Python UDF ハンドラーの例¶

前提条件¶

ファイルをコードで使用できるようにするには、Pythonハンドラーコードがステージ上のファイルを読み取る前に、次を実行する必要があります。

1. ハンドラーが使用可能なステージを作成します。

   外部ステージまたは内部ステージを使用できます。内部ステージを使用する場合、呼び出し元権限のストアドプロシージャを作成する予定であれば、これをユーザーステージにすることができます。それ以外の場合は、名前付きステージを使用する必要があります。Snowflakeは現在、UDF 依存関係のためにテーブルステージを使用してハンドラーコードを格納することはサポートしていません。

   ステージの作成の詳細については、 CREATE STAGE をご参照ください。内部ステージ型の選択の詳細については、 ローカルファイルに対する内部ステージの選択 をご参照ください。

   ユースケースに応じて、ステージに対する適切な権限を以下のロールに割り当てる必要があります。

   ユースケース	ロール
   UDF または所有者権限ストアドプロシージャ	実行中の UDF またはストアドプロシージャを所有するロール。
   呼び出し元権限ストアドプロシージャ	ユーザーロール。

   See moreShow less

   Expand

   詳細については、 ユーザー定義関数の権限付与 をご参照ください。

2. コードが読み取るファイルをステージにコピーします。

   PUT コマンドを使用して、ローカルドライブから内部ステージにファイルをコピーできます。PUT を使用したファイルのステージングについては、 ローカルファイルシステムからのデータファイルのステージング をご参照ください。

   クラウドストレージサービスが提供するツールを使用して、ローカルドライブから外部ステージの場所にファイルをコピーできます。ヘルプについては、ご使用のクラウドストレージサービスのドキュメントをご参照ください。

インラインPythonハンドラーを使用した画像の知覚ハッシュの算出¶

この例では、 SnowflakeFile を使用して、一対のステージングされた画像ファイルを読み取り、 知覚ハッシュ （pHash）を使用して、各ファイルの画像の類似性を判定します。

mode の引数に rb を渡して入力モードをバイナリに指定し、画像のphash値を返す UDF を作成します。

CREATE OR REPLACE FUNCTION calc_phash(file_path STRING)
  RETURNS STRING
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.12'
  PACKAGES = ('snowflake-snowpark-python','imagehash','pillow')
  HANDLER = 'run'
AS $$
from PIL import Image
import imagehash
from snowflake.snowpark.files import SnowflakeFile

def run(file_path):
  with SnowflakeFile.open(file_path, 'rb') as f:
  return imagehash.average_hash(Image.open(f))
$$;

2つの画像のphash値間における距離を計算する2番目の UDF を作成します。

CREATE OR REPLACE FUNCTION calc_phash_distance(h1 STRING, h2 STRING)
  RETURNS INT
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.12'
  PACKAGES = ('imagehash')
  HANDLER = 'run'
AS $$
import imagehash

def run(h1, h2):
  return imagehash.hex_to_hash(h1) - imagehash.hex_to_hash(h2)
$$;

画像ファイルをステージし、ディレクトリテーブルを更新します。

PUT file:///tmp/image1.jpg @images AUTO_COMPRESS=FALSE;
PUT file:///tmp/image2.jpg @images AUTO_COMPRESS=FALSE;

ALTER STAGE images REFRESH;

UDFs を呼び出します。

SELECT
  calc_phash_distance(
    calc_phash(build_scoped_file_url(@images, 'image1.jpg')),
    calc_phash(build_scoped_file_url(@images, 'image2.jpg'))
  ) ;

UDTF を使用した CSV ファイルの処理¶

この例では、 SnowflakeFile を使って、 CSV ファイルの内容を抽出し、その行をテーブルにして返す UDTF を作成します。

インラインハンドラーを使用して UDTF を作成します。

CREATE FUNCTION parse_csv(file_path STRING)
  RETURNS TABLE (col1 STRING, col2 STRING, col3 STRING)
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.12'
  PACKAGES = ('snowflake-snowpark-python')
  HANDLER = 'csvparser'
AS $$
from snowflake.snowpark.files import SnowflakeFile

class csvparser:
  def process(self, stagefile):
    with SnowflakeFile.open(stagefile) as f:
      for line in f.readlines():
        lineStr = line.strip()
        row = lineStr.split(",")
        try:
          # Read the columns from the line.
          yield (row[1], row[0], row[2], )
        except:
          pass
$$;

CSV ファイルをステージし、ディレクトリテーブルを更新します。

PUT file:///tmp/sample.csv @data_stage AUTO_COMPRESS=FALSE;

ALTER STAGE data_stage REFRESH;

UDTF を呼び出し、ファイル URL を渡します。

SELECT * FROM TABLE(PARSE_CSV(build_scoped_file_url(@data_stage, 'sample.csv')));

複数ファイルの処理¶

ディレクトリテーブルの RELATIVE_PATH 列をハンドラーに渡すと、複数のファイルを読み取って処理することができます。RELATIVE_PATH 列の詳細については、 ディレクトリテーブルのクエリからの出力 をご参照ください。

UDF を呼び出して、複数のファイルを処理します。

次の例では、 CREATE TABLE ステートメント内で UDF を呼び出し、ステージ上の各ファイルを処理し、その結果を新しいテーブルに格納します。

デモのために、例では以下のように想定しています。

• my_stage という名前のステージには、複数のテキストファイルがあります。

• 非構造化テキストに対して感情分析を行う、 get_sentiment という名前の既存の UDF があります。UDF は、テキストファイルへのパスを入力とし、感情を表す値を返します。

CopyExpand

CREATE OR REPLACE TABLE sentiment_results AS
SELECT
  relative_path
  , get_sentiment(build_scoped_file_url(@my_stage, relative_path)) AS sentiment
FROM directory(@my_stage);

Show lessSee more

Scroll to top

UDTF を呼び出して、複数のファイルを処理します。

次の例では、 UDTF という名前の parse_excel_udtf を呼び出します。この例では、ディレクトリテーブルから relative_path を my_excel_stage というステージに渡します。

CopyExpand

SELECT t.*
FROM directory(@my_stage) d,
TABLE(parse_excel_udtf(build_scoped_file_url(@my_excel_stage, relative_path)) t;

Show lessSee more

Scroll to top

ステージ URLs および URIs ハンドラーを使用したファイルの読み取り¶

SnowflakeFile を使用したファイルアクセスには、デフォルトでスコープ付き URLs が必要です。これにより、ファイルインジェクション攻撃に対して耐久性のあるコードになります。ただし、代わりにステージ URI、またはステージ URL を使用してファイルの場所を参照することはできます。そのためには、キーワード引数 require_scoped_url = False を使用して、 SnowflakeFile.open メソッドを呼び出す必要があります。

このオプションは、 UDF 所有者のみがアクセスできる URI を呼び出し元が提供できるようにする場合に便利です。たとえば、 UDF を所有していて、構成ファイルや機械学習モデルを読み取る場合、ファイルアクセスにステージ URI を使用することがあります。ユーザーの入力に基づいて作成されるファイルなど、予測不可能な名前を持つファイルを扱う場合は、このオプションを推奨しません。

この例では、機械学習モデルをファイルから読み取り、そのモデルを関数で使用して、感情分析のための自然言語処理を実行します。この例では、 open を require_scoped_url = False で呼び出します。ファイルの場所の形式両方（ステージ URI とステージ URL）で、 UDF 所有者はモデルファイルにアクセスできる必要があります。

インラインハンドラーを使用して UDF を作成します。

CREATE OR REPLACE FUNCTION extract_sentiment(input_data STRING)
  RETURNS STRING
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.12'
  PACKAGES = ('snowflake-snowpark-python','scikit-learn')
  HANDLER = 'run'
AS $$
from snowflake.snowpark.files import SnowflakeFile
from sklearn.linear_model import SGDClassifier
import pickle

def run(input_data):
  model_file = '@models/NLP_model.pickle'
  # Specify 'mode = rb' to open the file in binary mode.
  with SnowflakeFile.open(model_file, 'rb', require_scoped_url = False) as f:
    model = pickle.load(f)
    return model.predict([input_data])[0]
$$;

モデルファイルをステージし、ディレクトリテーブルを更新します。

PUT file:///tmp/NLP_model.pickle @models AUTO_COMPRESS=FALSE;

ALTER STAGE models REFRESH;

また、 UDF にモデルのステージ URL を指定して、感情をを抽出することもできます。

たとえば、ステージ URL を使用して、ファイルを指定するインラインハンドラー付き UDF を作成します。

CREATE OR REPLACE FUNCTION extract_sentiment(input_data STRING)
  RETURNS STRING
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.12'
  PACKAGES = ('snowflake-snowpark-python','scikit-learn')
  HANDLER = 'run'
AS $$
from snowflake.snowpark.files import SnowflakeFile
from sklearn.linear_model import SGDClassifier
import pickle

def run(input_data):
  model_file = 'https://my_account/api/files/my_db/my_schema/models/NLP_model.pickle'
  # Specify 'rb' to open the file in binary mode.
  with SnowflakeFile.open(model_file, 'rb', require_scoped_url = False) as f:
    model = pickle.load(f)
    return model.predict([input_data])[0]
$$;

入力されたデータで UDF を呼び出します。

SELECT extract_sentiment('I am writing to express my interest in a recent posting made.');