Snowflake Notebooks 参数传递完整指南

如何构建生产级 Notebook 流水线，并在每种工作区类型和执行模式下进行参数传递。

所有场景的完整可运行代码均可在 GitHub 仓库 中找到。

本指南涵盖每个场景，并提供可复制到你自己账户中的可运行示例。

理解整体格局：

工作区类型

执行模式

这一区分对下面每一个模式都至关重要。

术语： 本指南使用 Code Bundle（原称 Notebook Project Object/NPO）。Snowflake 重命名了该对象；旧版 NOTEBOOK PROJECT SQL 语法仍受支持。参见 Snowflake Code Bundles。

场景 1：%run —— 内存中的共享变量

最适用于： 同一工作区中的顺序 Notebook。这是最简单的方法。

工作原理： %run 在 同一个 Python 进程中执行另一个 Notebook。所有变量、函数和导入都是共享的。

架构

发送方在其最后一个单元格中设置变量：

OUTPUT_TABLE = f'{TARGET_DB}.{TARGET_SCHEMA}.ORDER_SUMMARY'
ROW_COUNT = session.sql(f'SELECT COUNT(*) AS C FROM {OUTPUT_TABLE}').collect()[0]['C']
STATUS = 'SUCCESS'

编排器调用两者，并可在它们之间进行门控：

%run data_prep.ipynb

assert STATUS == 'SUCCESS', f"Data prep failed: {STATUS}"

%run report.ipynb

接收方直接读取变量——无需导入，无需解析：

print(f"Reading from: {OUTPUT_TABLE}")   # Already in memory
report_df = session.sql(f"SELECT * FROM {OUTPUT_TABLE}")

完整 Notebook 代码：GitHub 仓库 / scenario-1-run

场景 2：使用 SYSTEM$SET_RETURN_VALUE 的 Task DAG

最适用于： 作为独立生产单元运行的独立 Notebook。

每个 Notebook 都是自己的 Code Bundle。Task DAG 将它们串联起来。第一个任务设置返回值，第二个任务读取它并将其作为 ARGUMENTS 传递。

发送方 设置一个以竖线分隔的返回值（不是 JSON——参见陷阱 #4）：

return_payload = f"{OUTPUT_TABLE}|{ROW_COUNT}|{RUN_DATE}|{STATUS}"

try:
    session.sql(f"SELECT SYSTEM$SET_RETURN_VALUE('{return_payload}')").collect()
except Exception as e:
    if 'must be called from within a task' in str(e):
        print("Interactive mode - skipping")
    else:
        raise

Task DAG 读取并转发该值：

CREATE OR REPLACE TASK report_task
  WAREHOUSE = 'COMPUTE_WH'
  AFTER data_prep_task
AS
BEGIN
  LET predecessor_output VARCHAR := SYSTEM$GET_PREDECESSOR_RETURN_VALUE(
    'DATA_PREP_TASK'    -- Short name only! See Gotcha #5
  );
  EXECUTE CODE BUNDLE "MY_DB"."MY_SCHEMA"."REPORT_BUNDLE"
    ENTRYPOINT = 'report.ipynb'
    ARGUMENTS = (:predecessor_output);
END;

接收方 解析以竖线分隔的字符串：

user_args = [a for a in sys.argv[1:] if not a.startswith('-') and not a.startswith('/')]

parts = user_args[0].split('|')
source_table = parts[0]    # 'MY_DB.MY_SCHEMA.ORDER_SUMMARY'
row_count = parts[1]       # '5'
run_date = parts[2]        # '2026-09-30'
status = parts[3]          # 'SUCCESS'

变体：用于静态参数的 Task CONFIG

如果在创建任务时参数已知，请使用 CONFIG——它允许你在不重新创建任务的情况下更新参数：

CREATE OR REPLACE TASK nightly_run
  SCHEDULE = 'USING CRON 0 9 * * * America/Los_Angeles'
  CONFIG = $${"run_date": "2026-09-30", "target_db": "PROD_DB"}$$
AS
BEGIN
  LET run_date VARCHAR := SYSTEM$GET_TASK_GRAPH_CONFIG('run_date')::VARCHAR;
  LET target_db VARCHAR := SYSTEM$GET_TASK_GRAPH_CONFIG('target_db')::VARCHAR;
  -- ARGUMENTS accepts only literals or variable references, not expressions,
  -- so concatenate into a variable first
  LET args VARCHAR := :run_date || '|' || :target_db;
  EXECUTE CODE BUNDLE ... ARGUMENTS = (:args);
END;

-- Update without recreating the task
ALTER TASK nightly_run SET CONFIG = $${"run_date": "2026-10-31", "target_db": "STAGING_DB"}$$;

完整 Notebook + SQL 代码：GitHub 仓库 / scenario-2-task-dag

场景 3：多个前驱（扇入）

最适用于： 依赖多个上游 Notebook 输出的 Notebook。

对每个前驱调用一次 SYSTEM$GET_PREDECESSOR_RETURN_VALUE，并用双竖线分隔符组合：

CREATE OR REPLACE TASK report_task
  AFTER data_prep_task, enrichment_task, validation_task
AS
BEGIN
  LET prep_out VARCHAR := SYSTEM$GET_PREDECESSOR_RETURN_VALUE('DATA_PREP_TASK');
  LET enrich_out VARCHAR := SYSTEM$GET_PREDECESSOR_RETURN_VALUE('ENRICHMENT_TASK');
  LET valid_out VARCHAR := SYSTEM$GET_PREDECESSOR_RETURN_VALUE('VALIDATION_TASK');

  LET all_args VARCHAR := :prep_out || '||' || :enrich_out || '||' || :valid_out;
  EXECUTE CODE BUNDLE ... ARGUMENTS = (:all_args);
END;

Notebook 先按双竖线拆分，然后对每个前驱按单竖线拆分：

predecessor_outputs = user_args[0].split('||')
prep_parts = predecessor_outputs[0].split('|')       # TABLE|COUNT|DATE|STATUS
enrich_parts = predecessor_outputs[1].split('|')
valid_parts = predecessor_outputs[2].split('|')

完整 Notebook + SQL 代码：GitHub 仓库 / scenario-3-task-dag

场景 4：同一个 Notebook 跨不同 DAG

关键洞见：Notebook 与 DAG 解耦。每个任务向同一个 Code Bundle 传递不同的参数：

-- DAG 1
CREATE OR REPLACE TASK sales_report AFTER sales_prep AS
BEGIN
  LET val VARCHAR := SYSTEM$GET_PREDECESSOR_RETURN_VALUE('SALES_PREP');
  EXECUTE CODE BUNDLE "MY_DB"."MY_SCHEMA"."REPORT_BUNDLE"
    ENTRYPOINT = 'report.ipynb'
    ARGUMENTS = (:val);
END;

-- DAG 2 (same Code Bundle, different predecessor)
CREATE OR REPLACE TASK marketing_report AFTER marketing_prep AS
BEGIN
  LET val VARCHAR := SYSTEM$GET_PREDECESSOR_RETURN_VALUE('MARKETING_PREP');
  EXECUTE CODE BUNDLE "MY_DB"."MY_SCHEMA"."REPORT_BUNDLE"
    ENTRYPOINT = 'report.ipynb'
    ARGUMENTS = (:val);
END;

Notebook 只是读取 sys.argv[1]——它并不知道也不关心是哪个 DAG 调用了它。

完整 Notebook + SQL 代码：GitHub 仓库 / scenario-4-task-dag

经验教训

这些是让我们耗费数小时的问题。其他地方均无相关文档。

1. get_active_session() 在无头模式下失败

无头执行可能创建多个会话，导致 SnowparkSessionException: More than one active session。

from snowflake.snowpark import session as snowpark_session
try:
    session = get_active_session()
except Exception:
    session = next(iter(snowpark_session._active_sessions))

2. SYSTEM$SET_RETURN_VALUE 仅在 Task 内有效

从工作表或交互式运行时抛出：Function SYSTEM$SET_RETURN_VALUE must be called from within a task。

始终用 try/except 包裹 以便 Notebook 在两种模式下都能工作。

3. JSON 在 ARGUMENTS 中会被破坏

如果你设置 JSON 返回值：

SYSTEM$SET_RETURN_VALUE('{"output_table": "MY_TABLE"}')

ARGUMENTS 解析器会剥离双引号。Notebook 收到的是 {output_table: MY_TABLE}——不是有效的 JSON。

修复：改用竖线分隔格式而非 JSON：

# Sender
return_payload = f"{table_name}|{row_count}|{run_date}|{status}"

# Receiver
parts = sys.argv[1].split('|')

4. SYSTEM$GET_PREDECESSOR_RETURN_VALUE 需要短任务名

-- FAILS:
SYSTEM$GET_PREDECESSOR_RETURN_VALUE('MY_DB.MY_SCHEMA.DATA_PREP_TASK')

-- WORKS:
SYSTEM$GET_PREDECESSOR_RETURN_VALUE('DATA_PREP_TASK')

5. 编辑 Notebook 不会更新 Code Bundle

在工作区更改代码后，你必须点击“Deploy Changes”（日历图标 > Deploy Changes）。否则任务会继续运行旧版本。

快速参考

监控

在 Snowflake Notebook 之间传递参数：完整指南 最初发布于 Snowflake Builders Blog: Data Engineers, App Developers, AI, & Data Science（Medium），人们正在那里通过高亮并回应这个故事来继续对话。