Snowflake ML 作业¶

将 Python 函数作为 Snowflake ML 作业运行¶

使用 Function Dispatch 在 Snowflake 的计算资源上使用 @remote 装饰器远程运行单个 Python 函数。

使用 @remote 装饰器，您可以执行以下操作：

• 序列化函数及其依赖项。

• 将其上传到指定 Snowflake 暂存区。

• 在 ML 容器运行时中执行。

以下示例 Python 代码使用 @remote 装饰器提交 Snowflake ML 作业。请注意，Snowpark Session 必填，请参阅 先决条件。

from snowflake.ml.jobs import remote

@remote("MY_COMPUTE_POOL", stage_name="payload_stage", session=session)
def train_model(data_table: str):
  # Provide your ML code here, including imports and function calls
  ...

job = train_model("my_training_data")

调用 @remote 装饰函数会返回一个 Snowflake MLJob 对象，该对象可用于管理和监控作业执行。有关更多信息，请参阅 Snowflake ML 作业管理。

将 Python 文件作为 Snowflake ML 作业运行¶

在 Snowflake 计算资源上运行 Python 文件或项目目录。这在以下情况下很有用：

• 您的 ML 项目很复杂，有多个模块和依赖项。

• 您希望保持本地开发与生产代码之间的分离。

• 您需要运行使用命令行实参的脚本。

• 您正在处理不是专门为在 Snowflake 计算上执行而设计的现有 ML 项目。

Snowflake 作业 API 提供两种主要方法：

• submit_file()：用于运行单个 Python 文件

• submit_directory()：用于运行包含多个文件和资源的整个项目目录

两种方法都支持：

• 命令行实参传递

• 环境变量配置

• 自定义依赖项规范

• 通过 Snowflake 暂存区进行项目资产管理

File Dispatch 对于生产现有 ML 工作流程以及保持开发和执行环境之间的明确分离特别有用。

以下 Python 代码向 Snowflake ML 作业提交一个文件：

from snowflake.ml.jobs import submit_file

# Run a single file
job1 = submit_file(
  "train.py",
  "MY_COMPUTE_POOL",
  stage_name="payload_stage",
  args=["--data-table", "my_training_data"],
  session=session,
)

以下 Python 代码向 Snowflake ML 作业提交一个目录：

from snowflake.ml.jobs import submit_directory

# Run from a directory
job2 = submit_directory(
  "./ml_project/",
  "MY_COMPUTE_POOL",
  entrypoint="train.py",
  stage_name="payload_stage",
  session=session,
)

提交文件或目录会返回一个 Snowflake MLJob 对象，该对象可用于管理和监控任务执行。有关更多信息，请参阅 Snowflake ML 作业管理。

专用包源¶

Snowflake ML 作业还支持从 JFrog Artifactory 和 Sonatype Nexus 存储库等专用源加载包。这些源通常用于分发内部和专有包、保持对依赖版本的控制以及确保安全性/合规性。

要从专用源安装软件包，必须执行以下操作：

1. 创建网络规则以允许访问专用源 URL。

   1. 对于使用基本身份验证的源，您只需创建网络规则即可。

      CREATE OR REPLACE NETWORK RULE private_feed_nr
      MODE = EGRESS
      TYPE = HOST_PORT
      VALUE_LIST = ('<your-repo>.jfrog.io');
      

      Copy

   2. 要使用专用连接（即专用链接）配置对源的访问权限，请按照 使用专用连接的网络出口 中的步骤操作。

2. 使用网络规则创建外部访问集成。向将要提交作业的角色授予 EAI 使用权限。

   CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION private_feed_eai
   ALLOWED_NETWORK_RULES = (PRIVATE_FEED_NR)
   ENABLED = true;
   
   GRANT USAGE ON INTEGRATION private_feed_eai TO ROLE <role_name>;
   

   Copy

3. 提交作业时指定专有源 URL、外部访问集成和包

   # Option 1: Specify private feed URL in pip_requirements
   job = submit_file(
     "/path/to/script.py",
     compute_pool="MY_COMPUTE_POOL",
     stage_name="payload_stage",
     pip_requirements=[
       "--index-url=https://your.private.feed.url",
       "internal-package==1.2.3"
     ],
     external_access_integrations=["PRIVATE_FEED_EAI"]
   )
   

   Copy

   # Option 2: Specify private feed URL by environment variable
   job = submit_directory(
     "/path/to/code/",
     compute_pool="MY_COMPUTE_POOL",
     entrypoint="script.py",
     stage_name="payload_stage",
     pip_requirements=["internal-package==1.2.3"],
     external_access_integrations=["PRIVATE_FEED_EAI"],
     env_vars={'PIP_INDEX_URL': 'https://your.private.feed.url'},
   )
   

   Copy

如果您的专有源 URL 包含身份验证令牌等敏感信息，请通过创建 Snowflake 密钥管理 URL。使用 CREATE SECRET 创建一个密钥。在作业提交期间使用 :spec_overrides 实参配置密钥。

# Create secret for private feed URL with embedded auth token
feed_url = "<your-repo>.jfrog.io/artifactory/api/pypi/test-pypi/simple"
user = "<auth_user>"
token = "<auth_token>"
session.sql(f"""
CREATE SECRET IF NOT EXISTS PRIVATE_FEED_URL_SECRET
 TYPE = GENERIC_STRING
 SECRET_STRING = 'https://{auth_user}:{auth_token}@{feed_url}'
""").collect()

# Prepare service spec override for mounting secret into job execution
spec_overrides = {
 "spec": {
  "containers": [
    {
     "name": "main",  # Primary container name is always "main"
     "secrets": [
      {
        "snowflakeSecret": "PRIVATE_FEED_URL_SECRET",
        "envVarName": "PIP_INDEX_URL",
        "secretKeyRef": "secret_string"
      },
     ],
    }
  ]
 }
}

# Load private feed URL from secret (e.g. if URL includes auth token)
job = submit_file(
  "/path/to/script.py",
  compute_pool="MY_COMPUTE_POOL",
  stage_name="payload_stage",
  pip_requirements=[
    "internal-package==1.2.3"
  ],
  external_access_integrations=["PRIVATE_FEED_EAI"],
  spec_overrides=spec_overrides,
)

有关 container.secrets 的更多信息，请参阅 containers.secrets 字段。