Snowflake Code Bundles 实战指南（一）：Python

在 Snowflake 上运行代码的新方式

笔记本很好用。但在写了这么多年代码之后，我常常更愿意直接在我最喜欢的 IDE 中工作，享受自动 linting、良好的格式化和快速测试带来的好处。此外，基于文件的组织方式让配置环境、组合可复用模块、管理依赖以及编排更复杂的逻辑都变得更加容易。它们提供了将探索性工作转变为可靠、可重复的自动化工作负载所需的结构和灵活性。

如果这也是你的风格，那有个好消息。借助 Snowflake Code Bundles，你现在可以直接在 Snowflake 中执行你的代码文件：在本地开发和测试代码，然后在 Snowflake 计算上运行它。你的代码来到你的数据身边，无需将数据移出 Snowflake，这既提供了可扩展性，也让人安心。

本文是解释 Code Bundles 如何在 Snowflake 中工作的系列文章的第一篇，本篇聚焦于 Python 运行时。

在本文中，我们将涵盖：

• 什么是 Snowflake Code Bundle
• Code Bundle 可观测性
• 保留 Code Bundle 版本
• 高级 Code Bundle 模式
• 将 Code Bundles 调度为作业

什么是 Snowflake Code Bundle

Snowflake CODE BUNDLE 是一种新的 schema 级对象类型，它允许你将代码、依赖项和配置打包成一个可移植单元，该单元可以在本地测试并在 Snowflake 管理的计算上执行。你可以将文件上传到 stage，或将它们保存在 workspace （私有或共享）中。然后，使用唯一名称创建一个 CODE BUNDLE 对象。

CREATE OR REPLACE CODE BUNDLE dummy_project
    FROM @<database>.<schema>.<stage>/<path_to_project_directory>;

SHOW CODE BUNDLES;

或者，从 workspace 创建它（在我的示例中，workspace 名称是“Code Bundle Test”，由于名称中包含空格，限定它时需要用双引号括起来：

CREATE OR REPLACE CODE BUNDLE dummy_project
    FROM 'snow://workspace/"USER$".PUBLIC."Code Bundle Test"/versions/live/'

以上以 live/ 结尾的路径是 workspace 的根目录，要指定更低层级的路径，请将项目路径追加到其后。

一个 code bundle 对象映射到一个 特定路径，该路径可以是包含你所有项目文件的项目目录。你所有的项目文件及其依赖项都保持在一起，而你来定义入口点。因此它被称为 code bundle。你的文件树可能像这样：

my_project/
├── code_bundle.yml      # runtime config file
├── requirements.txt     # dependency
├── helper.py            # dependency
├── main.py              # entry point
└── README.md

你所有的计算需求都需要在一个 YAML 文件中配置，并且它必须命名为 code_bundle.yml ——

• bundle.type 标识工作负载模式：spark 表示 spark 工作负载，custom 表示其他类型的工作负载。
• compute.type 允许你选择理想的计算资源 ——选项包括 warehouse 或 compute_pool。
• bundle.language 指定运行时，从 python、java 或 scala 中选择。
• compute.options 提供更多执行细节，包括运行时版本、warehouse 或 pool，以及相关设置。

以下是一个入门版 code bundle 配置的示例：

bundle:
  type: custom
  compute_type: warehouse
  language: python
  compute_options:
    runtime_version: '3.12'

执行一个 code bundle 可以像下面这样简单：

EXECUTE CODE BUNDLE dummy_project
    ENTRYPOINT = 'main.py';

如果你选择在 warehouse 中执行，尤其是当 warehouse 已经预热时，你可以有效消除运行时启动时间。

Code Bundle 可观测性

有一件极其重要的事情要记住：你在 code bundle 中打算打印的任何内容都不会显示在 SnowSight UI 中。这与直接以交互方式执行 Python 文件有很大不同。旧的 print 方法仍然会被执行，使你的代码可以在不同环境之间移植，只是结果不会显示在任何地方。

为了更好地观测 code bundle 执行期间发生的事件，最好利用与 Snowflake 事件表紧密集成的日志记录能力。

1. 日志记录 / 事件表

日志可观测性是 Snowflake 中最有用的功能之一。在 Python 中，它支持用于遥测消息传递的标准 logging 库。结果最终会进入你选择的事件表（账户默认事件表或你的自定义数据库级事件表）。

CREATE EVENT TABLE <database>.<schema>.code_bundle_events;

ALTER DATABASE <database>
    SET EVENT_TABLE = <database>.<schema>.code_bundle_events;

-- requires MODIFY LOG LEVEL ON ACCOUNT grant
ALTER DATABASE <database>
    SET LOG_LEVEL = 'INFO';

在入口文件中包含以下代码片段或类似内容，重新创建 code bundle，然后执行代码：

# main.py
import logging
import platform

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

if __name__ == "__main__":

    version = str(platform.python_version())
    logger.info(f"Python Version: {version}")

    if not version.startswith('3.12'):
        logger.error("Wrong Version!")
        raise Exception("XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX")

要查询事件表，请遵循标准 Snowflake 事件表格式来定位记录（记录可能需要几分钟才会显示）：

SELECT 
    TIMESTAMP AS event_ts,
    RESOURCE_ATTRIBUTES:"snow.query.id"::STRING AS query_id,
    RECORD:"severity_text"::STRING AS log_level,
    VALUE::STRING AS log_value
FROM code_bundle_events
WHERE TIMESTAMP >= DATEADD('hour', -1, CURRENT_TIMESTAMP())
AND RECORD_TYPE = 'LOG';

2. Code Bundle 执行历史记录

SNOWFLAKE.INFORMATION_SCHEMA.CODE_BUNDLE_HISTORY 是一个表函数，返回 code bundles 的执行历史记录。它提供运行级元数据，包括 bundle 标识、入口点、状态、时间、运行时和计算类型、执行标识符以及返回值。你可以应用过滤器，例如 bundle 名称、时间范围、状态、语言等。

SELECT * FROM TABLE(SNOWFLAKE.INFORMATION_SCHEMA.CODE_BUNDLE_HISTORY(
    DATABASE => '<database>',
    SCHEMA   => '<schema>',
    BUNDLE_NAME => 'DUMMY_PROJECT',
    RESULT_LIMIT => 20
));

这里是文档，说明如何查询 CODE_BUNDLE_HISTORY 表。

3. Code Bundle 执行历史记录 UI

如果你更喜欢上述执行历史记录数据的可视化仪表板视图，它也可以在 SnowSight UI 中随时使用。打开左侧导航栏，在 Catalog > Explorer 下 > 选择 database > 选择 schema > 然后选择你需要观测的 code bundle。浏览右侧的选项卡以打开其 Run History页面。该页面以表格格式显示历史记录，支持过滤，并直接提供指向执行详情、日志或查询历史的链接。

因为我在前面的步骤中明确要求你重新创建 code bundle，所以表函数结果和 SnowSight UI 中的执行历史记录都只返回它被重新创建之后的执行记录。为了不丢失历史状态和洞察，我们可以改为向现有 code bundle 对象使用 ADD VERSION，如下所述。

保留 Code Bundle 版本

基于文件的自动化程序的另一个好处是，你可以更轻松地进行版本控制。你不仅可以在本地 IDE 中或使用 git 服务轻松比较代码差异，还可以直接在 Snowflake 中跟踪 code bundle 对象的版本。Code Bundle 版本是 bundle 源文件的一个不可变快照。创建初始 bundle，然后使用 ALTER CODE BUNDLE ... ADD VERSION FROM ... 添加新的快照

ALTER CODE BUNDLE dummy_project
ADD VERSION FROM @<database>.<schema>.<stage>/<new_path>;

或者从 workspace 添加新版本：

ALTER CODE BUNDLE dummy_project
ADD VERSION FROM
  'snow://workspace/"USER$".PUBLIC.<workspace>/versions/live/<new_path>';

保留所有历史版本会带来更多好处。你可以轻松地将更改部署到同一个代码捆绑包对象，而不会破坏任何现有工作流，同时保留项目文件的每个版本的副本，直到你准备好丢弃它们。只需像这样组织你的项目文件夹：

<stage_name>/<project_folder>/V1         # initial version
<stage_name>/<project_folder>/V2         # new version
<stage_name>/<project_folder>/v20260808  # to be very specific

这样，你就能对过去的版本保持完全的可见性，从而在出现任何问题时都能非常轻松地恢复到之前可用的版本。要查看代码捆绑包的当前版本和路径，只需使用：

DESCRIBE CODE BUNDLE dummy_project;

请注意，即使你没有对文件做任何更改，并且从完全相同的路径进行部署，只要你使用 ADD VERSION 方法，你的代码捆绑包版本号元数据仍然会递增。

高级代码捆绑包模式

1. 使用容器运行时

有时 Snowflake 用户会选择计算池作为运行时，因为容器让他们能够使用隔离且可扩展的应用程序环境。你可以打包代码所需的库和运行时，在需要时使用 GPU，并运行大量的 Python 处理，而无需强制将所有内容都放入仓库执行中。当工作负载超出以 SQL 为中心的模式时，它提供了灵活性。

要在代码捆绑包中使用计算池运行时，需要先配置一个计算池，然后我们可以稍微修改 code_bundle.yml 文件以包含计算池配置。请注意，仍然需要指定一个查询仓库，用于执行查询。

bundle:
  type: custom
  compute_type: compute_pool
  language: python

  compute_options:
    compute_pool: <compute pool name>
    query_warehouse: <warehouse name>
    runtime_version: 'V2.5-CPU-PY3.12'

截至本文撰写之日，代码捆绑包仅在使用容器运行时时向账户级事件表发送遥测数据。

2. 在运行时自定义运行时

有时你可能需要在执行时动态控制运行时规格，而不是硬编码它。代码捆绑包允许你在 EXECUTE 语句中使用一个简单的关键字 WITH SPECIFICATION，以字符串字面量的形式提供运行时要求，这将在执行时完全覆盖你项目文件夹中任何现有的 YAML 文件。这在测试新环境兼容性时特别有用。

EXECUTE CODE BUNDLE dummy_project
    ENTRYPOINT = 'main.py'
    WITH SPECIFICATION
$$
bundle:
  type: custom
  compute_type: warehouse
  language: python

compute_options:
    runtime_version: '3.13'
$$;

这次执行如预期般失败，因为我们在代码中加入了一个自定义异常，当 Python 版本不是 3.12 时就会抛出错误。该过程失败了，但信息通过错误日志被捕获。

3. 灵活的参数解析

argparse 是我最喜欢的用于解析输入参数的 Python 库。在你的入口点文件中将 main.py 更新为类似这样：

# main.py
import argparse
...

if __name__ == "__main__":
    ...
    parser = argparse.ArgumentParser(description="Demo Code Bundles")
    parser.add_argument("-d", "--report-date", type=str, help="YYYY-MM-DD")
    parser.add_argument("-f", "--backfill", action="store_true")
    args = parser.parse_args()

    logger.info(f"{vars(args)}")
    ...

在常规 Python 运行时中，你会这样做：

/usr/bin/python3 /home/user/my_project/main.py -d 2026-08-01 -f

要在代码捆绑包执行中包含输入参数，你需要像下面这样将它们作为元组提供：

EXECUTE CODE BUNDLE dummy_project
    ENTRYPOINT = 'main.py'
    ARGUMENTS = ('--report-date', '2026-08-01', '--backfill');

查询事件表显示输入参数已成功传入运行时：

4. 使用自定义 Python 包

有时运行时的默认 Python 包并不是你的应用程序所需的版本。当你需要特定库版本时，在 code_bundle.yml 下添加一个 properties.requirements_file 条目，然后在该文件中列出锁定版本的依赖项。

以 jellyfish 为例，修改你的主文件以使用它：

if __name__ == "__main__":

    import importlib.metadata
    import jellyfish

    version = importlib.metadata.version("jellyfish")
    logger.info(f"jellyfish version: {version}")

因为 jellyfish 不在默认运行时中，尝试执行此代码捆绑包将无法工作。将以下内容添加到一个名为 requirements.txt 的文件中

# requirements.txt
snowpark-connect[jdk]==1.42.0

然后将这两行添加到 code_bundle.yml 中：

  # code_bundle.yml
...
  properties:
    requirements_file: requirements.txt

结果被成功捕获到事件表中：

5. 暂存区挂载存储路径

当代码或库期望使用文件系统路径（例如 open()）而不是 Snowflake 特有的文件 API 时，暂存区挂载就很有用。你可以使用 SnowflakeFile 读取文件，但写回暂存区需要路径解析；@stage/path 不是本地路径。目前，计算池上的暂存区挂载是可读写的。在仓库上，它们是只读的。

bundle:
  ...
  stage_mounts:
    input_data:
      stage_url: "@<database>.<schema>/<project_dir_path>"
      mount_path: "/mnt/data"

然后你的代码就可以正常访问这些文件：

with open("/mnt/data/file.csv") as f:
    contents = f.read()

6. 匿名代码捆绑包执行

匿名代码捆绑包执行直接从暂存区或工作区运行源代码，而不创建命名的代码捆绑包对象。它便于实验、版本化工件以及没有对象生命周期需要管理的一次性作业。

匿名代码捆绑包执行也有一些限制，例如数据库/模式沿袭较弱、遥测路由较差、可发现性较低，以及治理和历史方面的优势有限。

-- anonymous code bundle execution
EXECUTE CODE BUNDLE FROM '@<stage_name>/<temp_dir>'
  ENTRYPOINT = 'main.py';

EXECUTE CODE BUNDLE 
FROM 'snow://workspace/"USER$".PUBLIC.<workspace>/versions/live/'
  ENTRYPOINT = 'main.py';

将代码捆绑包作为作业调度

为了让你的代码捆绑包按计划自动运行，只需将它包含在一个 Snowflake TASK 中：

CREATE OR REPLACE TASK <database>.<schema>.<task>
  WAREHOUSE = MY_WAREHOUSE
  SCHEDULE = 'USING CRON 0 2 * * * UTC'
AS
  EXECUTE CODE BUNDLE <database>.<schema>.<code_bundle>
    ENTRYPOINT = 'main.py';

ALTER TASK <database>.<schema>.<task> RESUME;

请注意，你通常需要为你的 TASK 对象定义一个计算资源，并为你的 CODE BUNDLE 对象定义一个计算资源。这是两个不同的概念，它们不一定必须相同。如果你为任务本身选择 WAREHOUSE 计算，你的代码捆绑包可以从会话中固有地推断出仓库资源，你不需要再次定义它。

还有什么？你可以将这些任务链接在一起，以实现更复杂的编排。要更好地了解 Snowflake 任务 DAG 功能，请访问我的文章 Orchestrate Airflow-Style Task DAGs in Snowflake。

感谢你阅读本文！请尝试自己运行这些代码示例，如果遇到任何问题，请随时联系我。如果你希望将来看到类似的内容，请告诉我你正面临哪些挑战，或者正在寻找什么解决方案。让我们一起学习！

A Practical Guide to Snowflake Code Bundles — Part 1: Python 最初发布于 Snowflake Builders Blog: Data Engineers, App Developers, AI, & Data Science 在 Medium 上，人们通过点赞和回应这个故事来继续对话。