Snowpark Submit リファレンス¶

Snowpark Submit を使用すると、なじみのあるSparkセマンティクスを使用して、Snowflakeで非インタラクティブなバッチ指向のSparkワークロードを実行できます。

注釈

snowpark-submit は、spark-submit と同じ機能の多くをサポートしています。ただし、SnowflakeでSparkワークロードを実行する場合に必要ないため、一部の機能は省略されています。

構文¶

snowpark-submit
  --name <application_name>
  --exclude-packages <package_to_exclude> [, <package_to_exclude>, ...]
  --py-files <files_to_place_on_path>
  --conf <spark_config_property=value> [<spark_config_property=value> ...]
  --properties-file <path_to_properties_file>
  --help, -h
  --verbose, -v
  --version
  --account <snowflake_account>
  --user <snowflake_user>
  --authenticator <snowflake_authenticator>
  --token-file-path <snowflake_token_file_path>
  --password <snowflake_password>
  --role <snowflake_role>
  --host <snowflake_host>
  --database <snowflake_database_name>
  --schema <snowflake_schema_name>
  --warehouse <snowflake_warehouse_name>
  --compute-pool <snowflake_compute_pool>
  --comment <comment>
  --snowflake-stage <snowflake_stage>
  --external-access-integrations <snowflake_external_access_integrations> [, ...]
  --snowflake-log-level <snowflake_log_level>
  --snowflake-workload-name <snowflake_workload_name>
  --snowflake-connection-name <snowflake_connection_name>
  --snowflake-grpc-max-message-size <message_size>
  --snowflake-grpc-max-metadata-size <metadata_size>
  --workload-status
  --display-logs
  --wait-for-completion
  --jars <jar_files> [, <jar_files>, ...]
  --scala-version <scala_version>
  <application.jar | application.py> [<application_arguments>]

引数¶

application.jar | application.py: アプリケーションと依存関係を含むファイルへのパス。
[application arguments]: アプリケーションのメインメソッドに渡されるアプリケーション固有の引数。

オプション¶

--class CLASS_NAME: アプリケーションのメインクラス（JavaおよびScalaアプリケーションの場合）アプリケーション JAR でメインクラスが指定されていない場合、このオプションは必須です。
--conf [PROP=VALUEPROP=VALUE ...]: 任意のSpark構成プロパティ。
--exclude-packages [EXCLUDE_PACKAGES ...]: --packages で指定されている依存関係を解決する際に依存関係の衝突を回避するため除外する、groupId:artifactId ペアのコンマ区切りリスト。
--help, -h: ヘルプメッセージを表示して終了します。
--jars JAR: 含める .jar ファイルのコンマ区切りリスト。クラス検索が使用されていない場合は、ワークロード JAR 自体も含まれる場合があります。

注釈

ファイルはクラスパスに自動的に含まれないため、 addArtifact を使用して明示的に登録する必要があります。

--name NAME アプリケーションの名前。

--properties-file FILE: 追加プロパティの読み込み元であるファイルへのパス。指定しない場合は、conf/spark-defaults.confが検索されます。
--py-files PY_FILES: Pythonアプリ用に PYTHONPATH に配置する .zip ファイル、.egg ファイルまたは .py ファイルのコンマ区切りリスト。
--verbose, -v: 追加のデバッグ出力を出力します。
--version: 現在のSparkのバージョンを出力します。

Snowflake固有のオプション¶

--account SNOWFLAKE_ACCOUNT: 使用するSnowflakeアカウント。指定されている場合は、connections.toml ファイルのアカウントを上書きします。
--authenticator SNOWFLAKE_AUTHENTICATOR: Snowflakeでのログインの認証方式。指定されている場合は、connections.toml ファイルの認証コードを上書きします。指定しない場合のデフォルトはユーザーパスワード認証コードです。
--comment COMMENT: ワークロードに関連付けられたメッセージ。Snowflakeでワークロードを識別するために使用できます。
--compute-pool SNOWFLAKE_COMPUTE_POOL: 指定されたワークロードを実行するためのSnowflakeコンピューティングプール。指定されている場合は、connections.toml ファイルのコンピューティングプールを上書きします。
--database SNOWFLAKE_DATABASE_NAME: セッションで使用されるSnowflakeデータベース。指定されている場合は、connections.toml ファイルのデータベースを上書きします。
--display-logs: --workload-status が指定されている場合にアプリケーションログをコンソールに出力するかどうか。
--external-access-integrations [SNOWFLAKE_EXTERNAL_ACCESS_INTEGRATIONS ...]: ワークロードに必要なSnowflake外部アクセス統合。
--host SNOWFLAKE_HOST: Snowflakeのデプロイ用のホスト。指定されている場合は、connections.toml ファイルのホストを上書きします。
--password SNOWFLAKE_PASSWORD: Snowflakeユーザーのパスワード。指定されている場合は、connections.toml ファイルのパスワードを上書きします。
--requirements-file REQUIREMENTS_FILE: ワークロードを実行する前にインストールするPythonパッケージの依存関係を含む requirements.txt ファイルへのパス。PyPI の外部アクセス統合が必要です。--snowflake-stage パラメーターも指定した場合、このパラメーターは機能しません。
--role SNOWFLAKE_ROLE: 使用するSnowflakeロール。指定されている場合は、connections.toml ファイルのロールを上書きします。
--schema SNOWFLAKE_SCHEMA_NAME: セッションで使用するSnowflakeスキーマ。指定されている場合は、connections.toml ファイルのスキーマを上書きします。
--snowflake-connection-name SNOWFLAKE_CONNECTION_NAME: 基本構成として使用する connections.toml ファイルの接続の名前。コマンドライン引数は、connections.toml ファイルの値を上書きします。
--snowflake-grpc-max-message-size MESSAGE_SIZE: Snowpark Submit での gRPC 通信の最大メッセージサイズ（バイト単位）。
--snowflake-grpc-max-metadata-size METADATA_SIZE: Snowpark Submit での gRPC 通信の最大メタデータサイズ（バイト単位）。
--snowflake-log-level SNOWFLAKE_LOG_LEVEL: Snowflakeイベントテーブルのログレベル---'INFO'、'ERROR'、'NONE'`（デフォルト: :code:'INFO'`）。
--snowflake-stage SNOWFLAKE_STAGE: ワークロードファイルがアップロードされるSnowflakeステージ。
--snowflake-workload-name SNOWFLAKE_WORKLOAD_NAME: Snowflakeで実行されるワークロードの名前。
--token-file-path SNOWFLAKE_TOKEN_FILE_PATH: Snowflakeの OAuth トークンを含むファイルへのパス。指定されている場合は、connections.toml ファイル内のトークンファイルのパスを上書きします。
--user SNOWFLAKE_USER: 使用するSnowflakeユーザー。指定されている場合は、connections.toml ファイルのユーザーを上書きします。
--wait-for-completion: クラスターモードでは、指定されている場合は、ブロックモードでワークロードを実行し、完了を待ちます。
--warehouse SNOWFLAKE_WAREHOUSE_NAME: セッションで使用するSnowflakeウェアハウス。指定されている場合は、connections.toml ファイルのウェアハウスを上書きします。
--wheel-files WHEEL_FILES: Pythonワークロードを実行する前にインストールする.whlファイルのコンマ区切りリスト。PyPI では利用できないプライベート依存関係に使用します。
--workload-status: ワークロードの詳細ステータスを出力します。
--scala-version SCALA_VERSION: 使用するScalaバージョン。2.12 または 2.13。デフォルト値は 2.12 です。

一般的なオプションの例¶

アプリケーションのデプロイ¶

SnowflakeのSnowpark Container Services（SPCS）は、Sparkアプリケーションを実行するための主要なインフラストラクチャです。SPCS コンピューティングプールを事前に作成しておく必要があります。

Pythonの基本アプリケーション¶

クラスターモードで基本的なPythonアプリケーションをデプロイするには:

snowpark-submit \
  --snowflake-workload-name MY_PYTHON_JOB \
  --snowflake-connection-name MY_CONNECTION_CONFIG_NAME
  app.py arg1 arg2

Scalaの基本アプリケーション¶

Scalaの基本アプリケーションをクラスターモードで展開するには、次のようなコマンドを使用します。

snowpark-submit \
  --class com.example.MainClass \
  --snowflake-workload-name MY_SCALA_JOB \
  --snowflake-connection-name MY_CONNECTION_CONFIG_NAME
  app.jar arg1 arg2

アプリケーション JAR でメインクラスを指定した場合は、 --class オプションを省略できます。

Scalaバージョンの指定¶

Scala 2.12と2.13がサポートされており、デフォルトバージョンは2.12です。アプリケーションがScala 2.13で構築されている場合は、 CLI およびスクリプトでScalaバージョンを指定する必要があります。

snowpark-submit \
  --scala-version 2.13 \
  --snowflake-workload-name MY_SCALA_JOB \
  --snowflake-connection-name MY_CONNECTION_CONFIG_NAME
  app.jar arg1 arg2

アプリケーションコードでScalaバージョンを指定する必要があります。

// Directly in the session builder
val spark = SparkSession.builder()
  .remote("sc://localhost:15002")
  .config("snowpark.connect.scala.version", "2.13")
  .getOrCreate()

// Via session configuration
spark.conf.set("snowpark.connect.scala.version", "2.13")

認証¶

Snowpark Submit には、Snowflakeで認証するためのさまざまなメソッドが用意されています。少なくとも1つのメソッドを使用する必要があります。接続プロファイルと直接認証は、併用または個別に使用できます。接続プロファイルに対応するフィールドも存在する場合は、コマンドラインオプションによって該当するフィールドが上書きされます。

接続プロファイル¶

事前構成されたSnowflake接続プロファイルを使用するには:

snowpark-submit \
  --snowflake-connection-name my_connection \
  --snowflake-workload-name MY_JOB \
  app.py

直接認証¶

ユーザー名とパスワード¶

コマンドで認証の詳細情報を直接提示するには:

snowpark-submit \
  --host myhost \
  --account myaccount \
  --user myuser \
  --password mypassword \
  --role myrole \
  --snowflake-workload-name MY_JOB \
  app.py

OAuth¶

OAuth トークンを使用して認証するには:

snowpark-submit \
  --host myhost \
  --account myaccount \
  --authenticator oauth \
  --token-file-path /path/to/token.txt \
  --snowflake-workload-name MY_JOB \
  --compute-pool MY_COMPUTE_POOL \
  app.py

Snowflakeリソース¶

ジョブのSnowflakeデータベース、スキーマ、ウェアハウス、およびコンピューティングプールを指定するには:

snowpark-submit \
  --database MY_DB \
  --schema MY_SCHEMA \
  --warehouse MY_WH \
  --snowflake-workload-name MY_JOB \
  --snowflake-connection-name MY_CONNECTION \
  app.py

Snowflakeステージ¶

Snowpark Submit を使用して、Snowflakeステージにファイルを直接保存してアクセスすることができます。

Snowflakeステージのファイルを使用してジョブを送信するには:

snowpark-submit \
  --snowflake-stage @my_stage \
  --snowflake-workload-name MY_JOB \
  --snowflake-connection-name MY_CONNECTION \
  @my_stage/app.py

依存関係の管理¶

アプリケーションの依存関係を管理することができます。

Pythonの依存関係¶

アプリケーションに必要な追加のPythonファイルまたはアーカイブを指定するには:

snowpark-submit \
  --py-files dependencies.zip,module.py \
  --snowflake-workload-name MY_PYTHON_JOB \
  --snowflake-connection-name MY_CONNECTION \
  app.py

JavaまたはScalaの依存関係¶

JavaまたはScalaアプリケーション用の外部 JAR ファイルを含めるには:

snowpark-submit \
  --jars dep1.jar,dep2.jar \
  --snowflake-workload-name MY_SCALA_JOB \
  --snowflake-connection-name MY_CONNECTION \
  --compute-pool MY_COMPUTE_POOL \
  app.jar

--jars オプションを使用して依存関係を追加する場合は、 addArtifact を使用して明示的に登録する必要があります。

spark.addArtifact("dep1.jar")
spark.addArtifact("dep2.jar")

ステージングされた JAR ファイルを使用することもできます。これらのファイルは、 --jars オプションで指定する必要はありません。

spark.conf.set("snowpark.connect.udf.java.imports", "[@mystage/dep1.jar, @mystage/dep2.jar]")

モニタリングと制御¶

Snowpark Submit ジョブを効率的にモニタリングおよび制御できます。

ジョブの完了待ち¶

デフォルトでは、Snowpark Submit はジョブを開始し、直ちに結果を返します。ブロッキングモードで実行し、ジョブの終了を待つには:

snowpark-submit \
  --snowflake-connection-name my_connection \
  --snowflake-workload-name MY_JOB \
  --wait-for-completion \
  app.py

wait-for-completion フラグは、ジョブが完了（正常または失敗で）するまでコマンドがブロックするように指示し、定期的なステータスの更新を示します。これは、Apache Airflowを使用する場合など、他のタスクを進める前にジョブが完了することを確認する必要があるワークフローに有効です。

ワークロードステータスの確認¶

ワークロードのステータス（実行中または完了）を確認します。

snowpark-submit --snowflake-connection-name my_connection --snowflake-workload-name MY_JOB --workload-status

このコマンドは、ワークロードに関する次の情報を返します。

現在の状態（DEPLOYING、RUNNING、SUCCEEDED、FAILED）
開始時間と期間
サービスの詳細

アプリケーションログの表示¶

ワークロードステータスとともに詳細ログを表示するには:

snowpark-submit --snowflake-connection-name my_connection --snowflake-workload-name MY_JOB --workload-status --display-logs

display-logs フラグは、アプリケーションの出力ログを取得してコンソールに表示します。これらのログを使用して、以下のタスクを実行できます。

アプリケーションエラーのデバッグ
実行状況のモニタリング
アプリケーション出力を表示する

注釈

ログを取得する準備が整うまでに、数秒から1分までの若干のレイテンシが発生します。イベントテーブルを使用してログデータを格納しない場合、ログは5分以内などの短期間保持されます。

高度な構成¶

高度な構成によって Snowpark Submit ジョブを微調整します。

外部アクセス統合¶

Sparkアプリケーションから外部サービスに接続します。

snowpark-submit \
  --external-access-integrations "MY_NETWORK_RULE,MY_STORAGE_INTEGRATION" \
  --snowflake-workload-name MY_JOB \
  --snowflake-connection-name my_connection \
  app.py

ログレベルの構成¶

Snowflakeイベントテーブルへのアプリケーションのログレベルを制御します。

snowpark-submit \
  --snowflake-log-level INFO \
  --snowflake-workload-name MY_JOB \
  --snowflake-connection-name MY_CONNECTION \
  app.py

--snowflake-log-levelのオプション:INFO、ERROR、NONE。

ジョブコンテキストの追加¶

Snowflakeでワークロードをより簡単に識別できるように、説明的なコメントを追加します。

snowpark-submit \
  --comment "Daily data processing job" \
  --snowflake-workload-name MY_JOB \
  --snowflake-connection-name my_connection \
  app.py