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>]

Copy

实参¶

application.jar | application.py: 包含应用程序和依赖项的文件路径。
[application arguments]: 传递给应用程序主方法的特定应用程序实参。

选项¶

--class CLASS_NAME: Your application's main class (for Java and Scala applications). This option is required if the main class is not specified in the application JAR.
--conf [PROP=VALUEPROP=VALUE ...]: 任意 Spark 配置属性。
--exclude-packages [EXCLUDE_PACKAGES ...]: 以逗号分隔的 groupId:artifactId 对列表，用于在解析 --packages 中提供的依赖关系时将其排除，以避免依赖关系冲突。
--help, -h: 显示帮助消息并退出。
--jars JAR: Comma-separated list of .jar files to include. This can include the workload JAR itself, if a class finder is not used.

备注

The files are not automatically included in the classpath and need to be explicitly registered using addArtifact.

--name NAME Name of your application.

--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: Path to a requirements.txt file containing Python package dependencies to install before running the workload. Requires external access integration for PyPI. This parameter will not function if you also specify the --snowflake-stage parameter.
--role SNOWFLAKE_ROLE: 要使用的 Snowflake 角色如果已指定，则替换 connections.toml 文件中的角色。
--schema SNOWFLAKE_SCHEMA_NAME: 要在会话中使用的 Snowflake 架构。如果已指定，则替换 connections.toml 文件中的架构。
--snowflake-connection-name SNOWFLAKE_CONNECTION_NAME: Name of the connection in connections.toml file to use as the base configuration. Command-line arguments override any values from the connections.toml file.
--snowflake-grpc-max-message-size MESSAGE_SIZE: Maximum message size, in bytes, for gRPC communication in Snowpark Submit.
--snowflake-grpc-max-metadata-size METADATA_SIZE: Maximum metadata size, in bytes, for gRPC communication in Snowpark Submit.
--snowflake-log-level SNOWFLAKE_LOG_LEVEL: Snowflake 事件表的日志级别 – 'INFO'、'ERROR'、'NONE'。（默认值：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 version to use. Can be 2.12 or 2.13. The default value is 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

Copy

Basic Scala application¶

To deploy a basic Scala application in cluster mode, use a command such as the following:

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

Copy

You can omit the --class option if you specified the main class in the application JAR.

Specifying the Scala version¶

Scala 2.12 and 2.13 are supported, with the default version being 2.12. If the application is built with Scala 2.13, the Scala version must be specified in the CLI and the script.

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

Copy

You must specify the Scala version in the application code:

// 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")

Copy

身份验证¶

Snowpark Submit 提供了使用 Snowflake 进行身份验证的各种方法。必须使用至少一种方法。连接配置文件和直接身份验证可以一起使用，也可以单独使用。当命令行选项也存在时，它会替换连接配置文件中的相应字段。

连接配置文件¶

要使用预先配置的 Snowflake 连接配置文件，请执行以下操作：

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

Copy

直接身份验证¶

用户名和密码¶

要直接在命令中提供身份验证详细信息，请执行以下操作：

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

Copy

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

Copy

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

Copy

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

Copy

依赖项管理¶

您可以管理应用程序的依赖项。

Python 依赖项¶

要指定应用程序所需的其他 Python 文件或存档，请执行以下操作：

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

Copy

Java or Scala dependencies¶

To include external JAR files for Java or Scala applications:

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

Copy

When you add dependencies via the --jars option, you must explicitly register them using addArtifact.

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

Copy

It's also possible to use staged JAR files. These files do not have to be specified in the --jars option.

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

Copy

监控和控制¶

您可以有效地监控和控制 Snowpark Submit 作业。

等待作业完成¶

默认情况下，Snowpark Submit 会启动作业并立即返回。要在阻塞模式下运行并等待作业完成，请执行以下操作：

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

Copy

wait-for-completion 标志使命令阻塞直到作业完成（成功或失败），显示定期状态更新。这对于需要在继续执行其他任务（例如使用 Apache Airflow 时）前确保任务完成的工作流程很有用。

检查工作负载状态¶

检查工作负载的状态（正在运行或已完成）。

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

Copy

此命令返回有关工作负载的以下信息：

当前状态（DEPLOYING、RUNNING、SUCCEEDED、FAILED）
开始时间和持续时间
服务详细信息

查看应用程序日志¶

要查看详细日志以及工作负载状态，请执行以下操作：

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

Copy

display-logs 标志将提取应用程序的输出日志并将其打印到控制台。使用这些日志，您可以执行以下任务：

调试应用程序错误
监控执行进度
查看应用程序输出

备注

日志准备好提取的延迟很小，从几秒钟到一分钟不等。当不使用事件表存储日志数据时，日志仅保留很短的时间，例如五分钟或更短。

高级配置¶

使用高级配置微调 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

Copy

日志记录级别配置¶

控制应用程序在 Snowflake 事件表中的日志记录级别：

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

Copy

--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

Copy