PythonワークシートでSnowparkコードを記述する

Pythonワークシートに Snowpark コードを記述して、 Snowsight でSnowpark Pythonを使用してデータを処理します。Pythonワークシートにコードを記述することで、依存ライブラリをインストールする必要なくSnowflakeで開発とテストを実行できます。

Pythonワークシートで開発するには、次の手順を実行します。

  1. Snowflakeでロールとパッケージを準備します

  2. 開発用のワークシートを設定します

  3. PythonワークシートでSnowparkコードを記述します

  4. Pythonワークシートを実行します

たとえば、Snowflakeのステージまたはデータベースオブジェクトからデータを抽出し、データを変換し、変換されたデータをSnowflakeに保存するコードをPythonワークシートに記述できます。その後、 そのコードをストアドプロシージャとして展開し、データパイプラインを構築できます。すべて、Snowflakeから移動することなく行えます。

このトピックの内容:

Pythonのワークシートについて

Pythonワークシートを使用すると、 Snowsight でSnowpark Pythonを使用して、データの操作と変換を実行できます。 Snowflake Anacondaチャネルにリストされているサードパーティパッケージ を使用したり、ステージから独自のPythonファイルをインポートしてスクリプトで使用したりできます。

Pythonワークシートを実行した後、返された結果と出力をスクリプトで確認します。結果は、コードに応じて、文字列、バリアント、またはテーブルとして表示されます。 Pythonワークシートの実行 をご参照ください。

注釈

Pythonワークシートはローカル開発環境ではなくSnowflake内で実行されるため、 session.add_import を使用してPythonコードが依存するファイルを追加したり、 session.add_packages または session.add_requirements を使用してPythonコードで使用する必要があるパッケージを追加したりすることはできません。代わりに、これらのファイルをステージに追加して、コードで参照します。 Snowsightを使用したファイルのステージング をご参照ください。

Pythonワークシートには次の制限があります。

  • WARN より低いログレベルは、デフォルトではPythonワークシートの Output に表示されません。出力に下位レベルのメッセージをログするには、 logging モジュールなどのログライブラリを使用して、ログに記録されるメッセージのレベルを設定します。

  • ワークシート内のブレークポイントやPythonコードの一部のみの実行はサポートされていません。

  • 画像やウェブページはサポートされていません。Pythonコードによって生成された画像またはウェブページは、Pythonワークシートに表示できません。

  • PythonワークシートはデフォルトでPython 3.11を使用しますが、 Packages でサポートされている他のバージョンを選択することができます。

これらのオプションでサポートが必要な場合は、代わりにローカル開発環境の使用を検討してください。 Snowpark Pythonの開発環境の設定 をご参照ください。

Pythonワークシートの前提条件

Pythonワークシートを使用するには、次を行う必要があります。

Anaconda利用規約を確認して同意する

Snowflake内でAnacondaが提供するパッケージの使用を開始する前に、 外部製品規約 に同意する必要があります。

注釈

規約に同意するには、組織管理者(ORGADMIN ロールを使用)である必要があります。Snowflake アカウントで規約に同意する必要があるのは一度だけです。 アカウントの ORGADMIN ロールの有効化 をご参照ください。

  1. Snowsight にサインインします。

  2. Admin » Billing & Terms を選択します。

  3. Anaconda セクションで、 Enable を選択します。

  4. Anaconda Packages ダイアログで、リンクをクリックして 外部製品規約ページ を確認します。

  5. 規約に同意する場合は、 Acknowledge & Continue を選択します。

利用規約に同意しようとしたときにエラーが表示された場合は、ユーザープロファイルに名、姓、または電子メールアドレスが含まれていない可能性があります。管理者ロールを持っている場合は、 ユーザープロファイルにユーザー詳細を追加する を参照して、 Snowsight を使用してプロファイルを更新してください。または、アカウント管理者に連絡して アカウントを更新してください

ステージからワークシートにPythonファイルを追加する

Snowflakeには、Pythonワークシートの Snowflake Anacondaチャネル のSnowparkパッケージが含まれます。

PythonワークシートでAnacondaに含まれていないPythonファイルまたはパッケージを使用する場合は、ファイルをSnowflakeの名前付きステージにアップロードしてから、Pythonワークシートのパッケージのリストに追加する必要があります。

ワークシートでPythonパッケージを使用するには、以下を実行します。

  1. Snowsight にサインインします。

  2. Projects » Worksheets を開きます。

  3. + » Python Worksheet を選択します。

  4. データベースとスキーマを選択します。

  5. Packages » Stage Packages を選択します。

  6. ステージ内のパッケージへのパスを入力します。

    • ワークシート用に選択したデータベースとスキーマにパッケージが配置されているステージが含まれている場合、非修飾名を使用してステージを参照できます。たとえば、 @YourStage/path/to/example_package.py です。

    • 別のデータベースおよびスキーマでステージを参照するには、ステージの名前を完全修飾します。たとえば、 @Database.Schema.Stage/path/to/other_package.py です。

  7. インストール済みパッケージのリストにパッケージを追加するには、 Import を選択します。

  8. コードで import ステートメントを使用して、Pythonワークシートでパッケージを使用できます。たとえば、 example_package.pyother_package.py のファイルからパッケージをインポートした後、 example_package ファイルから function という関数をインポートするために以下のコードを記述し、コードで使用するために other_package パッケージをインポートします。

    from example_package import function
    import other_package
    
    Copy

注釈

ワークシートに追加するパッケージは、そのワークシートでのみ使用できます。別のPythonワークシートで同じパッケージを使用する場合は、この手順を使用してパッケージをそのワークシートに追加します。

詳細については、 コードで依存関係を利用できるようにする方法 をご参照ください。

Pythonワークシートで開発を開始する

ワークシートを開いて開発環境を構成するには、次の手順を実行します。

  1. Snowsight にサインインします。

  2. Projects » Worksheets を開きます。

  3. + » Python Worksheet を選択します。

  4. データベースとスキーマを選択します。

  5. ワークシートの実行に使用するウェアハウスを選択します。ユーザーのデフォルトのウェアハウスがある場合は、事前に選択されています。

    Pythonワークシートでは、Python ッケージを読み込んでPythonコードを実行するために実行中のウェアハウスが必要です。

  6. (オプション) Packages を選択してPythonライブラリをインストールします。

    • Pythonワークシートには snowflake-snowpark-python パッケージが必要で、常にインストールされます。

    • numpy、pandas、requests、urllib3など、 Snowflake Anacondaチャネルにリストされている パッケージを検索します。ワークシートで使用するためにインストールするパッケージを選択し、オプションで Installed Packages のリストにあるデフォルトのパッケージバージョンを変更します。

    • Stage Packages を選択し、ステージとパッケージのファイルパスを指定してから、 Import を選択して、独自のパッケージとPythonファイルを追加します。 ステージからワークシートにPythonファイルを追加する をご参照ください。

    インストールしたパッケージは Installed Packages の下に表示されます。

  7. ワークシートにPythonライブラリをインストールした場合は、 import ステートメントをコードに追加して、インストールしたライブラリを使用します。

    たとえば、Pythonワークシート用にscikit-learnパッケージをインストールする場合は、コードの最初にそのパッケージ用の import ステートメントを追加します。

    import scikit-learn
    
    Copy
  8. サンプルPythonコードを実行して、構成を検証します。

エラーメッセージまたはコードからの戻り値が Results セクションに表示されます。ログメッセージを表示するには、 Output を選択します。 Pythonワークシートの実行 をご参照ください。

PythonワークシートでSnowparkコードを記述する

Pythonワークシートを使用して開発を開始する 手順に従った後、サンプルコードを独自のコードに置き換えることができます。

ハンドラー関数内にSnowpark Pythonコードを記述します。

import snowflake.snowpark as snowpark

def main(session: snowpark.Session):
    # your code goes here
Copy

デフォルトのハンドラー関数は main ですが、ワークシートの Settings で変更できます。アクティブなハンドラーはワークシートで強調表示されます。

ボイラープレートコードで提供される session オブジェクトを使用して、Snowpark API ライブラリでSnowflakeのデータにアクセスします。たとえば、テーブルの DataFrame を作成したり、SQL ステートメントを実行したりできます。 Python用Snowpark開発者ガイド をご参照ください。

入力するにしたがい、Pythonメソッド、定義済み変数、データベースオブジェクトなどのオートコンプリートが表示されます。一部のサードパーティパッケージまたはステージからインポートされたファイルのオートコンプリートは表示されません。Pythonワークシートには、構文の強調表示とメソッドパラメーターのガイダンスも含まれています。ワークシートの Settings でリンティングと行の折り返しを構成できます。

異なるデータ型の結果を返す

Pythonコードを記述するときは、コード内の return ステートメントによって返されるデータの型を考慮し、ワークシートが結果を返す方法を調整してください。デフォルトでは、プレースホルダーコードが DataFrame を返すため、Pythonワークシートの戻り値の型はTable()です。

Pythonコードが返す内容に応じて、ワークシートの設定を変更して、出力を別の方法で表示することができます。

  • ハンドラー関数が DataFrame を返す場合は、デフォルトの戻り値の型 Table() を使用します。

  • collect メソッドなどを使用して、ハンドラー関数が Row オブジェクトのリストを返す場合は、戻り値の型を Variant に変更します。

  • ハンドラー関数が return "Hello Python" などの文字列を返す場合、または文字列としてキャストする値を返す場合は、戻り値の型を String に変更します。

  • count メソッドなどでハンドラー関数が整数を返す場合は、 Variant または String の戻り型を使用します。

一部の DataFrame メソッドの戻り型に関する詳細については、 DataFrame を評価するアクションの実行 をご参照ください。

別の型の結果を返すようにワークシートの設定を更新するには、次の手順を実行します。

  1. Snowsight にサインインします。

  2. Projects » Worksheets を開きます。

  3. 結果をテーブルとして表示するPythonワークシートを開きます。

  4. ワークシートの実行に使用するウェアハウスを選択します。ユーザーのデフォルトのウェアハウスがある場合は、事前に選択されています。ウェアハウスが実行中であることを確認します。

  5. Settings を選択し、 Return type にはハンドラー関数によって返される型を選択します。

  6. Pythonワークシートを実行します。

  7. Results パネルで結果を確認します。

ハンドラー関数への追加引数の引き渡し

Pythonワークシートでは、単一の引数(Snowpark Session オブジェクト)を受け取るPython関数をワークシートのハンドラーとして指定すると、その関数をテストできます。Pythonワークシートで定義された関数はすべて session: snowpark.Session 引数を渡す必要があります。

追加の引数を渡す関数をテストするには、次を実行します。

  1. 引数を関数に追加します。

  2. Snowpark Session を渡す別の単一引数関数を定義します。この関数では、複数の引数を持つ関数を呼び出し、追加の引数に値を渡し、関数の値を返します。

    たとえば、パッケージ言語列でパッケージのテーブルをフィルターするSnowpark Pythonコードを記述するには、以下のコードを記述します。

    import snowflake.snowpark as snowpark
    from snowflake.snowpark.functions import col
    
    # Add parameters with optional type hints to the main handler function
    def main(session: snowpark.Session, language: str):
      # Your code goes here, inside the "main" handler.
      table_name = 'information_schema.packages'
      dataFrame = session.table(table_name).filter(col("language") == language)
    
      # Print a sample of the dataFrame to standard output
      dataFrame.show()
    
      # The return value appears in the Results tab
      return dataFrame
    
    # Add a second function to supply a value for the language parameter to validate that your main handler function runs.
    def test_language(session: snowpark.Session):
      return main(session, 'java')
    
    Copy

    この例では、 main 関数は複数引数の関数で、 test_language 関数は、渡された引数値を使用してコードが実行されることを検証するために使用される単一引数の関数です。

  3. ワークシートを実行し、コードが引数値で実行されることを検証するハンドラー関数として単一引数関数を設定します。

    この例では、ハンドラーを test_language 関数に変更し、 Run を選択します。ワークシート Settings でハンドラーを変更するか、ハンドラー関数の横にある Show actions ライトバルブを選択して Set function "test_language" as handler を選択します。

ストアドプロシージャとしてPythonワークシートを展開する と、メインハンドラー関数を選択し、ストアドプロシージャの引数とマップされた型を確認できます。

Pythonワークシートの実行

Pythonワークシートを作成 したら、 Run を選択してPythonワークシートを実行します。ワークシートを実行すると、Pythonワークシートのすべてのコードが実行されます。コードの部分実行または増分実行はサポートされていません。

注釈

Snowflake Anacondaチャネルにリストされている パッケージを使用していて、Anacondaの条件をまだ受け入れていない場合は、パッケージが見つからないというエラーが表示されることがあります。 Anacondaのサードパーティパッケージの使用 をご参照ください。

コードによって生成された出力を確認する

Pythonワークシートの Output パネルで、Pythonコードの標準出力(stdout)または標準エラー(stderr)メッセージを確認できます。

Output パネルでは、次のタイプの関数からの出力を確認できます。

  • print() など、コンソールに書き込む関数。

  • Snowpark Pythonの DataFrame クラスの show メソッドなど、DataFrame を出力する関数。

注釈

出力は、コードの実行時にストリームに表示されるのではなく、すべてのPythonプロセスの実行が終了した後に表示されます。

ログ出力は仮ステージに書き込まれ、次が true である場合にのみキャプチャされます。

  • ワークシートのデータベースとスキーマを選択している。

  • 選択したデータベースは共有から作成されたものではなかった場合。

  • 選択したデータベースとスキーマに対する USAGE 権限を持つロールを使用して、ワークシートを実行する。

Pythonワークシートのクエリ履歴を確認する

Pythonワークシートが Snowsight で実行されると、匿名のストアドプロシージャがコードを実行し、コード内のSnowparkコマンドを実行するクエリを生成します。

Snowsight の Query History ページを使用して、実行されたクエリを確認できます。 Snowsightを使用してクエリ履歴を確認する をご参照ください。

たとえば、ワークシートを実行した後、次の操作を実行して、実行されたクエリを確認できます。

  1. ワークシートの Results を確認します。

  2. ワークシートの Query Details で、 その他のオプション » Copy Query ID を選択します

  3. Projects » Worksheets を選択して、ワークシートのリストに戻ります。

  4. Monitoring » Query History を選択します。

  5. Query History ページで、Pythonワークシートのクエリのみを表示します。

    1. Filters を選択し、 Query ID オプションを有効にします。

    2. Pythonワークシートのクエリ ID を入力します。

    3. Apply Filters を選択します。

  6. ワークシートに対して実行されたクエリを確認します。

Pythonワークシートのコード例

Pythonワークシートを記述する と、名前付きステージからのデータ読み取りなどのデータ変換と操作タスクを実行できます。

その他の例は Snowpark PythonでのDataFramesの使用 でご確認ください。

例: 簡単なSnowparkプログラムを記述する

この例では、Snowpark Pythonプログラムを記述して、小さな範囲の数値を生成し、コードがSnowflakeで作成するテーブルにその範囲を記述するか、テーブルがすでに存在する場合はそれを上書きします。このコード例を実行するには、テーブルを追加するデータベーススキーマに対する CREATE TABLE 権限を持っている必要があります。

  1. Snowsight にサインインします。

  2. Projects » Worksheets を開きます。

  3. + » Python Worksheet を選択します。

  4. テーブルを追加するデータベースとスキーマを選択します。

  5. ワークシートの実行に使用するウェアハウスを選択します。ユーザーのデフォルトのウェアハウスがある場合は、事前に選択されています。ウェアハウスが実行中であることを確認します。

  6. SnowparkのPythonコードを main 関数の一部として記述します。

    import snowflake.snowpark as snowpark
    
    def main(session: snowpark.Session):
      tableName = "range_table"
      df_range = session.range(1, 10, 2).to_df('a')
      df_range.write.mode("overwrite").save_as_table(tableName)
      return tableName + " table successfully created"
    
    Copy
  7. Settings を選択し、 Return type については、ハンドラー関数によって返される型として String を選択します。

  8. コードを実行します。

例: Pythonワークシートでデータを変換する

この例では、 SNOWFLAKE データベースの ACCOUNT_USAGE スキーマにある TASK_HISTORY ビューのエントリを、スケジュールされた時間と状態によって集計し、集計された出力を aggregate_task_history テーブルに保存するPythonコードを記述します。

注釈

この例ではアカウントの使用状況データをクエリするため、次のロールを使用する必要があります。

  1. Snowsight にサインインします。

  2. Projects » Worksheets を開きます。

  3. + » Python Worksheet を選択します。

  4. テーブルを追加するデータベースとスキーマを選択します。

  5. ワークシートの実行に使用するウェアハウスを選択します。ユーザーのデフォルトのウェアハウスがある場合は、事前に選択されています。ウェアハウスが実行中であることを確認します。

  6. SnowparkのPythonコードを main 関数の一部として記述します。

    import snowflake.snowpark as snowpark
    from snowflake.snowpark.functions import col
    from snowflake.snowpark.dataframe_reader import *
    from snowflake.snowpark.functions import *
    
    def main(session: snowpark.Session):
    
      inputTableName = "snowflake.account_usage.task_history"
      outputTableName = "aggregate_task_history"
    
      df = session.table(inputTableName)
      df.filter(col("STATE") != "SKIPPED")\
        .group_by(("SCHEDULED_TIME"), "STATE").count()\
        .write.mode("overwrite").save_as_table(outputTableName)
      return outputTableName + " table successfully written"
    
    Copy
  7. Settings を選択し、 Return type については、ハンドラー関数によって返される型として String を選択します。

  8. コードを実行します。

Pythonワークシートでコードを実行した後、 SQL ワークシートを開き、テーブルをクエリすることができます。 ワークシートを使用したデータのクエリ をご参照ください。

例: Pythonワークシートを使用してステージからファイルを読み込む

Snowpark Pythonでは、ステージからファイルを読み込んで、その内容をSnowflakeのテーブルに書き込んだり、ビューとして保存したりすることができます。この例では、Pythonコードは db1.public.files という名前付きステージから従業員データ data_0_0_0.csv.gz を含む CSV フォーマットの圧縮ファイルのコンテンツを読み込み、 employees という名前のテーブルにコンテンツを書き込みます。

注釈

このコード例を実行するには、以下のロールを使用する必要があります。

  • コードで使用されるステージ、データベース、およびスキーマに対する USAGE 権限。

  • テーブルを追加するデータベーススキーマに対する CREATE TABLE 権限。

  1. Snowsight にサインインします。

  2. Projects » Worksheets を開きます。

  3. + » Python Worksheet を選択します。

  4. テーブルを追加するデータベースとスキーマを選択します。

  5. ワークシートの実行に使用するウェアハウスを選択します。ユーザーのデフォルトのウェアハウスがある場合は、事前に選択されています。ウェアハウスが実行中であることを確認します。

  6. SnowparkのPythonコードを main 関数の一部として記述します。

    import snowflake.snowpark as snowpark
    from snowflake.snowpark.types import *
    
    schema_for_file = StructType([
      StructField("name", StringType()),
      StructField("role", StringType())
    ])
    
    fileLocation = "@DB1.PUBLIC.FILES/data_0_0_0.csv.gz"
    outputTableName = "employees"
    
    def main(session: snowpark.Session):
      df_reader = session.read.schema(schema_for_file)
      df = df_reader.csv(fileLocation)
      df.write.mode("overwrite").save_as_table(outputTableName)
    
      return outputTableName + " table successfully written from stage"
    
    Copy
  7. Settings を選択し、 Return type については、ハンドラー関数によって返される型として String を選択します。

  8. コードを実行します。

Pythonワークシートでコードを実行した後、 SQL ワークシートを開き、テーブルをクエリすることができます。 ワークシートを使用したデータのクエリ をご参照ください。

Snowparkを使用したステージでのファイル操作の詳細については、 Snowpark PythonでのDataFramesの使用 をご参照ください。