Python을 사용하여 Snowflake 함수 및 저장 프로시저 관리하기¶

Python을 사용하여 Snowflake에서 사용자 정의 함수(UDFs)와 저장 프로시저를 관리할 수 있습니다. UDF 또는 프로시저를 생성하는 경우 지원되는 처리기 언어 중 하나로 해당 논리를 작성한 다음 Snowflake Python APIs 을 사용하여 생성합니다. UDFs 및 프로시저에 대한 자세한 내용은 함수와 프로시저로 Snowflake 확장하기 섹션을 참조하십시오.

전제 조건¶

이 항목의 예제에서는 Snowflake와 연결하고 Snowflake Python APIs 을 사용할 수 있는 Root 오브젝트를 생성하는 코드를 추가했다고 가정합니다.

예를 들어, 다음 코드는 구성 파일에 정의된 연결 매개 변수를 사용하여 Snowflake에 대한 연결을 생성합니다.

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

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

해당 코드에서는 결과 Session 오브젝트를 사용하여 API의 유형과 메서드를 사용하기 위해 Root 오브젝트를 생성합니다. 자세한 내용은 Snowflake Python APIs 을 사용하여 Snowflake에 연결 섹션을 참조하십시오.

사용자 정의 함수(UDFs) 관리하기¶

사용자 정의 함수(UDFs)를 관리할 수 있으며, 이를 통해 사용자가 직접 작성하여 시스템을 확장해 Snowflake에서 제공하는 기본 제공 시스템 정의 함수를 통해 제공되지 않는 작업을 수행할 수 있습니다. UDF를 생성한 후에는 여러 번 재사용할 수 있습니다. 자세한 내용은 사용자 정의 함수 개요 섹션을 참조하십시오.

참고

API를 사용하여 UDFs를 호출하는 기능은 현재 지원되지 않습니다.

Snowflake Python APIs 은 다음 두 가지 별개 유형의 UDFs를 나타냅니다.

UserDefinedFunction: 이름, 인자 목록, 반환 유형, 함수 정의와 같은 UDF의 속성을 표시합니다.
UserDefinedFunctionResource: 해당 UserDefinedFunction 오브젝트를 가져오고, UDF의 이름을 바꾸고, UDF를 삭제하는 데 사용할 수 있는 메서드를 노출합니다.

UDF 만들기¶

UDF를 생성하려면 먼저 UserDefinedFunction 오브젝트를 생성한 다음 API Root 오브젝트에서 UserDefinedFunctionCollection 오브젝트를 생성합니다. UserDefinedFunctionCollection.create 를 사용하여 Snowflake에 새 UDF를 추가합니다.

UDF를 생성할 때 지원되는 다음 언어 중 하나로 코드가 작성된 핸들러를 지정합니다.

Python¶

다음 예제의 코드는 지정된 인자, 반환 유형, 언어 및 UDF Python 정의와 함께 my_db 데이터베이스 및 my_schema 스키마에서 이름이 my_python_function 인 UDF를 나타내는 UserDefinedFunction 오브젝트를 생성합니다.

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.12", packages=[], handler="udf"),
    body="""
def udf():
    return {"key": "value"}
    """,
)

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

Java¶

다음 예제의 코드는 지정된 인자, 반환 유형, 언어 및 UDF Java 정의와 함께 my_db 데이터베이스 및 my_schema 스키마에서 이름이 my_java_function 인 UDF를 나타내는 UserDefinedFunction 오브젝트를 생성합니다.

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)

JavaScript¶

다음 예제의 코드는 지정된 인자, 반환 유형, 언어 및 UDF JavaScript 정의와 함께 my_db 데이터베이스 및 my_schema 스키마에서 이름이 my_javascript_function 인 UDF를 나타내는 UserDefinedFunction 오브젝트를 생성합니다.

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)

Scala¶

다음 예제의 코드는 지정된 인자, 반환 유형, 언어 및 UDF Scala 정의와 함께 my_db 데이터베이스 및 my_schema 스키마에서 이름이 my_scala_function 인 UDF를 나타내는 UserDefinedFunction 오브젝트를 생성합니다.

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)

SQL¶

다음 예제의 코드는 지정된 인자, 반환 유형, 언어 및 UDF SQL 정의와 함께 my_db 데이터베이스 및 my_schema 스키마에서 이름이 my_sql_function 인 UDF를 나타내는 UserDefinedFunction 오브젝트를 생성합니다.

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)

UDF 세부 정보 가져오기¶

UserDefinedFunction 오브젝트를 반환하는 UserDefinedFunctionResource.fetch 메서드를 호출하여 UDF에 대한 정보를 얻을 수 있습니다.

다음 예제의 코드는 my_db 데이터베이스의 my_javascript_function(DOUBLE) UDF와 my_schema 스키마에 대한 정보를 가져옵니다.

참고

UDF 리소스 오브젝트를 가져올 때는 UDFs가 오버로드될 수 있으므로 전체 서명(UDF 이름과 매개 변수 데이터 타입)을 지정해야 합니다.

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

UDFs 나열하기¶

UserDefinedFunction 오브젝트의 PagedIter 반복기를 반환하는 UserDefinedFunctionCollection.iter 메서드를 사용하여 UDFs를 나열할 수 있습니다.

다음 예제의 코드는 my_db 데이터베이스와 my_schema 스키마에서 이름이 my_java 로 시작하는 UDFs를 나열한 다음 각각의 이름을 출력합니다.

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)

UDF 이름 바꾸기¶

UserDefinedFunctionResource 오브젝트가 포함된 UDF의 이름을 바꿀 수 있습니다.

다음 예제의 코드는 my_db 데이터베이스와 my_schema 스키마에서 my_javascript_function(DOUBLE) UDF 리소스 오브젝트를 가져온 다음 UDF의 이름을 my_other_js_function 으로 변경하는 동시에 my_other_db 데이터베이스와 my_other_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"
)

UDF 삭제하기¶

UserDefinedFunctionResource 오브젝트가 포함된 UDF를 삭제할 수 있습니다.

다음 예제의 코드는 my_javascript_function(DOUBLE) UDF 리소스 오브젝트를 가져온 다음 UDF를 삭제합니다.

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

저장 프로시저 관리하기¶

SQL을 실행하는 프로시저 코드로 시스템을 확장하기 위해 작성할 수 있는 저장 프로시저를 관리할 수 있습니다. 저장 프로시저에서 프로그래밍 방식 구문을 사용하여 분기 및 루핑을 수행할 수 있습니다. 저장 프로시저를 생성한 후에는 여러 번 재사용할 수 있습니다. 자세한 내용은 저장 프로시저 개요 섹션을 참조하십시오.

Snowflake Python APIs 은 두 가지 유형이 있는 절차를 나타냅니다.

Procedure: 프로시저의 이름, 인자 목록, 반환 유형, 프로시저 정의와 같은 속성을 노출합니다.
ProcedureResource: 해당 Procedure 오브젝트를 가져오고, 프로시저를 호출하고, 프로시저를 삭제하는 데 사용할 수 있는 메서드를 노출합니다.

프로시저 만들기¶

프로시저를 생성하려면 먼저 Procedure 오브젝트를 생성한 다음 API Root 오브젝트에서 ProcedureCollection 오브젝트를 생성합니다. ProcedureCollection.create 를 사용하여 Snowflake에 새 프로시저를 추가합니다.

다음 예제의 코드는 지정된 인자, 반환 유형 및 SQL 프로시저 정의를 사용하여 my_db 데이터베이스 및 my_schema 스키마에서 이름이 my_procedure 인 프로시저를 나타내는 Procedure 오브젝트를 생성합니다.

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)

프로시저 호출하기¶

ProcedureResource 오브젝트를 사용하여 프로시저를 호출할 수 있습니다.

다음 예제의 코드는 my_procedure(VARCHAR) 프로시저 리소스 오브젝트를 가져와서 CallArgumentList 오브젝트를 생성한 다음 해당 인자 목록을 사용하여 프로시저를 호출합니다.

참고

프로시저 리소스 오브젝트를 가져올 때는 프로시저가 오버로드될 수 있으므로 전체 서명(프로시저 이름과 매개 변수 데이터 타입)을 지정해야 합니다.

from snowflake.core.procedure import CallArgument, CallArgumentList

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

프로시저 세부 정보 가져오기¶

Procedure 오브젝트를 반환하는 ProcedureResource.fetch 메서드를 호출하여 프로시저에 대한 정보를 얻을 수 있습니다.

다음 예제의 코드는 my_db 데이터베이스의 my_procedure(VARCHAR) 프로시저와 my_schema 스키마에 대한 정보를 가져옵니다.

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

프로시저 나열하기¶

Procedure 오브젝트의 PagedIter 반복기를 반환하는 ProcedureCollection.iter 메서드를 사용하여 프로시저를 나열할 수 있습니다.

다음 예제의 코드는 my_db 데이터베이스와 my_schema 스키마에서 이름이 my 로 시작하는 프로시저를 나열한 다음 각각의 이름을 인쇄합니다.

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

프로시저 삭제하기¶

ProcedureResource 오브젝트가 포함된 프로시저를 삭제할 수 있습니다.

다음 예제의 코드는 my_db 데이터베이스와 my_schema 스키마에서 my_procedure(VARCHAR) 프로시저 리소스 오브젝트를 가져온 다음 프로시저를 삭제합니다.

my_procedure_res = root.databases["my_db"].schemas["my_schema"].procedures["my_procedure(VARCHAR)"]
my_procedure_res.drop()