Verwalten von Snowflake-Funktionen und gespeicherten Prozeduren mit Python

Sie können Python verwenden, um benutzerdefinierte Funktionen (UDFs) und gespeicherte Prozeduren in Snowflake zu verwalten. Wenn Sie eine UDF oder eine Prozedur erstellen, schreiben Sie deren Logik in einer der unterstützten Handler-Sprachen und erstellen sie dann mit Snowflake Python APIs. Weitere Informationen über UDFs und Prozeduren finden Sie unter Erweitern von Snowflake um Funktionen und Prozeduren.

Voraussetzungen

Die Beispiele in diesem Thema gehen davon aus, dass Sie Code hinzugefügt haben, um eine Verbindung zu Snowflake herzustellen und ein Root-Objekt zu erstellen, von dem aus Sie die Snowflake Python APIs verwenden können.

Beispielsweise verwendet der folgende Code Verbindungsparameter, die in einer Konfigurationsdatei definiert sind, um eine Verbindung zu Snowflake zu erstellen:

from snowflake.core import Root
from snowflake.snowpark import Session

session = Session.builder.config("connection_name", "myconnection").create()
root = Root(session)
Copy

Unter Verwendung des resultierenden Session-Objekts erstellt der Code ein Root-Objekt, das die Typen und Methoden der API verwendet. Weitere Informationen dazu finden Sie unter Verbinden mit Snowflake mit dem Snowflake Python APIs.

Verwalten benutzerdefinierter Funktionen (UDFs)

Sie können benutzerdefinierte Funktionen (UDFs) verwalten, die Sie schreiben können, um das System zu erweitern, um Operationen auszuführen, die nicht über die integrierten, systemdefinierten Funktionen von Snowflake verfügbar sind. Nachdem Sie eine UDF erstellt haben, können Sie sie mehrfach wiederverwenden. Weitere Informationen dazu finden Sie unter Überblick über benutzerdefinierte Funktionen.

Bemerkung

Der Aufruf von UDFs über die API wird derzeit nicht unterstützt.

Die Snowflake Python APIs stellt UDFs mit zwei verschiedenen Typen dar:

  • UserDefinedFunction: Zeigt die Eigenschaften der UDF an, z. B. den Namen, die Liste der Argumente, den Rückgabetyp und die Funktionsdefinition.

  • UserDefinedFunctionResource: Stellt Methoden zur Verfügung, mit denen Sie ein entsprechendes UserDefinedFunction-Objekt abrufen, die UDF umbenennen und die UDF löschen können.

Erstellen einer UDF

Um ein UDF zu erstellen, legen Sie zunächst ein UserDefinedFunction-Objekt an und erstellen dann ein UserDefinedFunctionCollection-Objekt aus dem API Root-Objekt. Fügen Sie mit UserDefinedFunctionCollection.create die neue UDF zu Snowflake hinzu.

Wenn Sie eine UDF erstellen, geben Sie einen Handler an, dessen Code in einer der folgenden unterstützten Sprachen geschrieben ist.

Python

Der Code im folgenden Beispiel erstellt ein UserDefinedFunction-Objekt, das eine UDF mit dem Namen my_python_function in der my_db-Datenbank und dem my_schema-Schema mit den angegebenen Argumenten, dem Rückgabetyp, der Sprache und der UDF-Python-Definition darstellt:

from snowflake.core.user_defined_function import (
    PythonFunction,
    ReturnDataType,
    UserDefinedFunction
)

function_of_python = UserDefinedFunction(
    "my_python_function",
    arguments=[],
    return_type=ReturnDataType(datatype="VARIANT"),
    language_config=PythonFunction(runtime_version="3.9", packages=[], handler="udf"),
    body="""
def udf():
    return {"key": "value"}
    """,
)

root.databases["my_db"].schemas["my_schema"].user_defined_functions.create(function_of_python)
Copy

Java

Der Code im folgenden Beispiel erstellt ein UserDefinedFunction-Objekt, das eine UDF mit dem Namen my_java_function in der my_db-Datenbank und dem my_schema-Schema darstellt, mit den angegebenen Argumenten, dem Rückgabetyp, der Sprache und der UDF-Java- Definition:

from snowflake.core.user_defined_function import (
    Argument,
    JavaFunction,
    ReturnDataType,
    UserDefinedFunction
)

function_body = """
    class TestFunc {
        public static String echoVarchar(String x) {
            return x;
        }
    }
"""

function_of_java = UserDefinedFunction(
    name="my_java_function",
    arguments=[Argument(name="x", datatype="STRING")],
    return_type=ReturnDataType(datatype="VARCHAR", nullable=True),
    language_config=JavaFunction(
        handler="TestFunc.echoVarchar",
        runtime_version="11",
        target_path=target_path,
        packages=[],
        called_on_null_input=True,
        is_volatile=True,
    ),
    body=function_body,
    comment="test_comment",
)

root.databases["my_db"].schemas["my_schema"].user_defined_functions.create(function_of_java)
Copy

JavaScript

Der Code im folgenden Beispiel erstellt ein UserDefinedFunction-Objekt, das eine UDF mit dem Namen my_javascript_function in der my_db-Datenbank und dem my_schema-Schema mit den angegebenen Argumenten, dem Rückgabetyp, der Sprache und der UDF JavaScript-Definition darstellt:

from snowflake.core.user_defined_function import (
    Argument,
    ReturnDataType,
    JavaScriptFunction,
    UserDefinedFunction
)

function_body = """
    if (D <= 0) {
        return 1;
    } else {
        var result = 1;
        for (var i = 2; i <= D; i++) {
            result = result * i;
        }
        return result;
    }
"""

function_of_javascript = UserDefinedFunction(
    name="my_javascript_function",
    arguments=[Argument(name="d", datatype="DOUBLE")],
    return_type=ReturnDataType(datatype="DOUBLE"),
    language_config=JavaScriptFunction(),
    body=function_body,
)

root.databases["my_db"].schemas["my_schema"].user_defined_functions.create(function_of_javascript)
Copy

Scala

Der Code im folgenden Beispiel erstellt ein UserDefinedFunction-Objekt, das eine UDF mit dem Namen my_scala_function in der my_db-Datenbank und dem my_schema-Schema mit den angegebenen Argumenten, dem Rückgabetyp, der Sprache und der UDF-Scala-Definition darstellt:

from snowflake.core.user_defined_function import (
    Argument,
    ReturnDataType,
    ScalaFunction,
    UserDefinedFunction
)

function_body = """
    class Echo {
        def echoVarchar(x : String): String = {
            return x
        }
    }
"""

function_of_scala = UserDefinedFunction(
    name="my_scala_function",
    arguments=[Argument(name="x", datatype="VARCHAR")],
    return_type=ReturnDataType(datatype="VARCHAR"),
    language_config=ScalaFunction(
        runtime_version="2.12", handler="Echo.echoVarchar", target_path=target_path, packages=[]
    ),
    body=function_body,
    comment="test_comment",
)

root.databases["my_db"].schemas["my_schema"].user_defined_functions.create(function_of_scala)
Copy

SQL

Der Code im folgenden Beispiel erstellt ein UserDefinedFunction-Objekt, das eine UDF mit dem Namen my_sql_function in der my_db-Datenbank und dem my_schema-Schema mit den angegebenen Argumenten, dem Rückgabetyp, der Sprache und der UDF SQL-Definition darstellt:

from snowflake.core.user_defined_function import (
    ReturnDataType,
    SQLFunction,
    UserDefinedFunction
)

function_body = """3.141592654::FLOAT"""

function_of_sql = UserDefinedFunction(
    name="my_sql_function",
    arguments=[],
    return_type=ReturnDataType(datatype="FLOAT"),
    language_config=SQLFunction(),
    body=function_body,
)

root.databases["my_db"].schemas["my_schema"].user_defined_functions.create(function_of_sql)
Copy

Abrufen von UDF-Details

Sie können Informationen über eine UDF erhalten, indem Sie die UserDefinedFunctionResource.fetch-Methode aufrufen, die ein UserDefinedFunction-Objekt zurückgibt.

Der Code im folgenden Beispiel ruft Informationen über die my_javascript_function(DOUBLE) UDF in der my_db-Datenbank und das my_schema-Schema ab:

Bemerkung

Wenn Sie ein UDF-Ressourceobjekt abrufen, müssen Sie die vollständige Signatur (den UDF-Namen und ihre Parameterdatentypen) angeben, da UDFs überladen sein können.

my_udf = root.databases["my_db"].schemas["my_schema"].user_defined_functions["my_javascript_function(DOUBLE)"].fetch()
print(my_udf.to_dict())
Copy

Auflisten von UDFs

Sie können UDFs mit der UserDefinedFunctionCollection.iter-Methode auflisten, die einen PagedIter-Iterator von UserDefinedFunction-Objekten zurückgibt.

Der Code im folgenden Beispiel listet UDFs auf, deren Namen in der Datenbank my_db und im my_schema-Schema mit my_java beginnen und gibt den Namen jedes einzelnen aus:

udf_iter = root.databases["my_db"].schemas["my_schema"].user_defined_functions.iter(like="my_java%")
for udf_obj in udf_iter:
    print(udf_obj.name)
Copy

Umbenennen einer UDF.

Sie können eine UDF mit einem UserDefinedFunctionResource-Objekt umbenennen.

Der Code im folgenden Beispiel ruft das my_javascript_function(DOUBLE) UDF-Ressourcenobjekt in der my_db-Datenbank und im my_schema-Schema ab, benennt die UDF dann in my_other_js_function um und verschiebt die gleichzeitig in die my_other_db-Datenbank und das my_other_schema-Schema:

root.databases["my_db"].schemas["my_schema"].user_defined_functions["my_javascript_function(DOUBLE)"].rename(
    "my_other_js_function",
    target_database = "my_other_database",
    target_schema = "my_other_schema"
)
Copy

Löschen einer UDF

Sie können eine UDF mit einem UserDefinedFunctionResource-Objekt löschen.

Der Code im folgenden Beispiel ruft das my_javascript_function(DOUBLE) UDF-Ressourcenobjekt ab und löscht die UDF dann:

my_udf_res = root.databases["my_db"].schemas["my_schema"].user_defined_functions["my_javascript_function(DOUBLE)"]
my_udf_res.drop()
Copy

Verwalten von gespeicherten Prozeduren

Sie können gespeicherte Prozeduren verwalten, die Sie schreiben können, um das System mit prozeduralem Code zu erweitern, der SQL ausführt. In einer gespeicherten Prozedur können Sie Programmsteuerungskonstrukte verwenden, um Verzweigungen und Schleifen auszuführen. Nachdem Sie eine gespeicherte Prozedur erstellt haben, können Sie sie mehrfach wiederverwenden. Weitere Informationen dazu finden Sie unter Übersicht zu gespeicherten Prozeduren.

Die Snowflake Python APIs stellt Prozeduren mit zwei verschiedenen Typen dar:

  • Procedure: Zeigt die Eigenschaften einer Prozedur an, z. B. ihren Namen, die Liste der Argumente, den Rückgabetyp und die Prozedurdefinition.

  • ProcedureResource: Stellt Methoden zur Verfügung, mit denen Sie ein entsprechendes Procedure-Objekt abrufen, die Prozedur aufrufen und die Prozedur löschen können.

Erstellen einer Prozedur

Um eine Prozedur zu erstellen, erstellen Sie zunächst ein Procedure-Objekt und dann ein ProcedureCollection-Objekt aus dem API Root-Objekt. Fügen Sie mit ProcedureCollection.create die neue Prozedur zu Snowflake hinzu.

Der Code im folgenden Beispiel erstellt ein Procedure-Objekt, das eine Prozedur mit dem Namen my_procedure in der my_db-Datenbank und dem my_schema-Schema darstellt, mit den angegebenen Argumenten, dem Rückgabetyp und der SQL-Prozedurdefinition:

from snowflake.core.procedure import Argument, ColumnType, Procedure, ReturnTable, SQLFunction

procedure = Procedure(
    name="my_procedure",
    arguments=[Argument(name="id", datatype="VARCHAR")],
    return_type=ReturnTable(
        column_list=[
            ColumnType(name="id", datatype="NUMBER),
            ColumnType(name="price", datatype="NUMBER"),
        ]
    ),
    language_config=SQLFunction(),
    body="
        DECLARE
            res RESULTSET DEFAULT (SELECT * FROM invoices WHERE id = :id);
        BEGIN
            RETURN TABLE(res);
        END;
    ",
)

procedures = root.databases["my_db"].schemas["my_schema"].procedures
procedures.create(procedure)
Copy

Aufrufen einer Prozedur

Sie können eine Prozedur mit einem ProcedureResource-Objekt aufrufen.

Der Code im folgenden Beispiel ruft das Ressourcenobjekt der my_procedure(NUMBER, NUMBER)-Prozedur ab, erstellt ein CallArgumentList-Objekt und ruft mit dieser Liste von Argumenten dann die Prozedur auf.

Bemerkung

Wenn Sie ein Ressourceobjekt für eine Prozedur abrufen, müssen Sie die vollständige Signatur (den Namen der Prozedur und ihre Parameterdatentypen) angeben, da Prozeduren überladen sein können.

from snowflake.core.procedure import CallArgument, CallArgumentList

procedure_reference = root.databases["my_db"].schemas["my_schema"].procedures["my_procedure(NUMBER, NUMBER)"]
call_argument_list = CallArgumentList(call_arguments=[
    CallArgument(name="id", datatype="NUMBER", value=1),
])
procedure_reference.call(call_argument_list)
Copy

Abrufen von Prozedurdetails

Sie können Informationen über eine Prozedur erhalten, indem Sie die ProcedureResource.fetch-Methode aufrufen, die ein Procedure-Objekt zurückgibt.

Der Code im folgenden Beispiel holt Informationen über die my_procedure(NUMBER, NUMBER)-Prozedur aus der my_db-Datenbank und dem my_schema-Schema:

my_procedure = root.databases["my_db"].schemas["my_schema"].procedures["my_procedure(NUMBER, NUMBER)"].fetch()
print(my_procedure.to_dict())
Copy

Auflisten von Prozeduren

Sie können Prozeduren mit der ProcedureCollection.iter-Methode auflisten, die einen PagedIter-Iterator von Procedure-Objekten zurückgibt.

Der Code im folgenden Beispiel listet Prozeduren auf, deren Name mit my in der my_db-Datenbank und dem my_schema-Schema beginnt, und gibt dann den Namen jeder Prozedur aus:

procedure_iter = root.databases["my_db"].schemas["my_schema"].procedures.iter(like="my%")
for procedure_obj in procedure_iter:
    print(procedure_obj.name)
Copy

Löschen einer Prozedur

Sie können eine Prozedur mit einem ProcedureResource-Objekt löschen.

Der Code im folgenden Beispiel ruft das my_procedure(NUMBER, NUMBER)-Ressourcenobjekt der Prozedur in der my_db-Datenbank und dem my_schema-Schema ab und löscht die Prozedur dann.

my_procedure_res = root.databases["my_db"].schemas["my_schema"].procedures["my_procedure(NUMBER, NUMBER)"]
my_procedure_res.drop()
Copy
Sprache: Deutsch