PythonでDataFramesのストアドプロシージャを作成¶

概要¶

Snowparkを使用すると、カスタムラムダと関数のストアドプロシージャを作成でき、これらのストアドプロシージャを呼び出してDataFrameのデータを処理できます。

現在のセッション内にのみ存在するストアドプロシージャ（一時的なストアドプロシージャ）と、他のセッションで使用できるストアドプロシージャ（永続的なストアドプロシージャ）を作成できます。

ストアドプロシージャでのAnacondaのサードパーティパッケージの使用¶

Pythonストアドプロシージャを作成するときにインストールするAnacondaパッケージを指定できます。Snowflakeウェアハウス内でPythonストアドプロシージャを呼び出すと、ユーザーに代わってAnacondaパッケージをシームレスにインストールし、仮想ウェアハウスにキャッシュします。ベストプラクティス、利用可能なパッケージの表示方法、ローカル開発環境のセットアップ方法の詳細については、 サードパーティパッケージの使用 をご参照ください。

session.add_packages を使用して、セッションレベルでパッケージを追加します。

このコード例は、パッケージをインポートしてそのバージョンを返す方法を示しています。

>>> import pandas as pd
>>> import snowflake.snowpark
>>> import xgboost as xgb
>>> from snowflake.snowpark.functions import sproc

>>> session.add_packages("snowflake-snowpark-python", "pandas", "xgboost==1.5.0")

>>> @sproc
... def compute(session: snowflake.snowpark.Session) -> list:
...   return [pd.__version__, xgb.__version__]

session.add_requirements を使用して、 要件ファイル でパッケージを指定することもできます。

>>> session.add_requirements("mydir/requirements.txt")

ストアドプロシージャレベルのパッケージを追加して、以前に追加した可能性のあるセッションレベルのパッケージを上書きできます。

>>> import pandas as pd
>>> import snowflake.snowpark
>>> import xgboost as xgb
>>> from snowflake.snowpark.functions import sproc

>>> @sproc(packages=["snowflake-snowpark-python", "pandas", "xgboost==1.5.0"])
... def compute(session: snowflake.snowpark.Session) -> list:
...    return [pd.__version__, xgb.__version__]

匿名ストアドプロシージャの作成¶

匿名ストアドプロシージャを作成するには、次のいずれかを実行できます。

• snowflake.snowpark.functions モジュールで sproc 関数を呼び出し、匿名関数の定義を渡します。

• StoredProcedureRegistration クラスで register メソッドを呼び出し、匿名関数の定義を渡します。 StoredProcedureRegistration クラスの属性またはメソッドにアクセスするには、 Session クラスの sproc プロパティを呼び出します。

匿名ストアドプロシージャの例を次に示します。

>>> from snowflake.snowpark.functions import sproc
>>> from snowflake.snowpark.types import IntegerType

>>> add_one = sproc(lambda session, x: session.sql(f"select {x} + 1").collect()[0][0], return_type=IntegerType(), input_types=[IntegerType()], packages=["snowflake-snowpark-python"])

名前付きストアドプロシージャの作成と登録¶

ストアドプロシージャを名前で呼び出す場合（たとえば、 Session オブジェクトの call 関数を使用）、名前付きストアドプロシージャを作成して登録できます。これを行うには、次のいずれかを実行できます。

• snowflake.snowpark.functions モジュールで sproc 関数を呼び出し、 name 引数と匿名関数の定義を渡します。

• StoredProcedureRegistration クラスで register メソッドを呼び出し、 name 引数と匿名関数の定義を渡します。 StoredProcedureRegistration クラスの属性またはメソッドにアクセスするには、 Session クラスの sproc プロパティを呼び出します。

register を呼び出すか、または sproc が現在のセッションで使用できる一時ストアドプロシージャを作成します。

永続的なストアドプロシージャを作成するには、 register メソッドまたは sproc 関数を呼び出し、 is_permanent 引数を True に設定します。永続的なストアドプロシージャを作成するときは、 stage_location 引数を、Snowparkが使用するPythonコネクタがストアドプロシージャとその依存関係のPythonファイルをアップロードするステージの場所にも設定する必要があります。

名前付きの一時ストアドプロシージャを登録する方法の例を次に示します。

>>> from snowflake.snowpark.functions import sproc
>>> from snowflake.snowpark.types import IntegerType

>>> add_one = sproc(lambda session, x: session.sql(f"select {x} + 1").collect()[0][0],
return_type=IntegerType(), input_types=[IntegerType()], name="my_sproc", replace=True,
packages=["snowflake-snowpark-python"])

is_permanent 引数を True に設定して、名前付きの永続ストアドプロシージャを登録する方法の例を次に示します。

>>> import snowflake.snowpark
>>> from snowflake.snowpark.functions import sproc

>>> @sproc(name="minus_one", is_permanent=True, stage_location="@my_stage", replace=True, packages=["snowflake-snowpark-python"])
... def minus_one(session: snowflake.snowpark.Session, x: int) -> int:
...  return session.sql(f"select {x} - 1").collect()[0][0]

これらのストアドプロシージャが呼び出される例を次に示します。

>>> add_one(1)
2
>>> session.call("minus_one", 1)
0
>>> session.sql("call minus_one(1)").collect()
[Row(MINUS_ONE(1)=0)]

ストアドプロシージャからのファイルの読み取り¶

前述のように、Snowparkライブラリはサーバーにストアドプロシージャをアップロードして実行します。ストアドプロシージャがファイルからデータを読み取る必要がある場合は、ファイルがストアドプロシージャとともにアップロードされていることを確認する必要があります。

ファイルを読み取るようにストアドプロシージャを設定するには、

1. ファイルが依存関係であることを指定します。これにより、ファイルがサーバーにアップロードされます。これは、UDFsの場合と同じ方法で行われます。詳細については、 UDF の依存関係の指定 をご参照ください。

   例:

   >>> # Import a file from your local machine as a dependency.
   >>> session.add_import("/<path>/my_file.txt")
   
   >>> # Or import a file that you uploaded to a stage as a dependency.
   >>> session.add_import("@my_stage/<path>/my_file.txt")
   

   Copy

2. ストアドプロシージャで、ファイルを読み取ります。

   >>> def read_file(name: str) -> str:
   ...    import sys
   ...    IMPORT_DIRECTORY_NAME = "snowflake_import_directory"
   ...    import_dir = sys._xoptions[IMPORT_DIRECTORY_NAME]
   ...
   ...    with open(import_dir + 'my_file.txt', 'r') as file:
   ...        return file.read()
   

   Copy