如何获取Delta Live Table的pipeline_id与update_id并写入数据？

参数对应关系说明 #

DLT的pipeline_id和update_id与Databricks Job的参数直接等价：

• pipeline_id 对应 Job ID，可通过dbutils.widgets.get("job_id")获取
• update_id 对应 Job Run ID，可通过dbutils.widgets.get("run_id")获取

你之前关注的job_id和run_id就是这两个参数的直接映射，无需额外转换。

直接在DLT表定义中注入元数据 #

完全不需要用event_hooks（该方式更适合监听管道生命周期事件，而非给数据行打标记），直接在表的构建逻辑中获取参数并添加字段即可，这是最直接高效的方案：

Python示例

import dlt
from pyspark.sql.functions import lit

@dlt.table(name="enriched_data_table")
def build_enriched_table():
    # 获取DLT管道元数据参数
    pipeline_id = dbutils.widgets.get("job_id")
    update_id = dbutils.widgets.get("run_id")
    
    # 读取源数据并追加元数据字段
    source_df = spark.read.table("raw_source_table")
    return source_df.withColumn("pipeline_id", lit(pipeline_id)) \
                    .withColumn("update_id", lit(update_id))

SQL示例

如果使用DLT SQL语法，可直接通过变量引用参数：

CREATE OR REFRESH LIVE TABLE enriched_data_table
AS SELECT
  *,
  '$job_id' AS pipeline_id,
  '$run_id' AS update_id
FROM LIVE.raw_source_table;

为什么不推荐event_hooks？ #

event_hooks的设计初衷是响应管道的生命周期事件（如启动、成功、失败），用来记录管道级日志或触发外部动作。用它给每一行数据打标记属于间接实现，不仅增加逻辑复杂度，还可能带来不必要的性能开销，远不如直接在表定义中注入参数的方案合理。

内容的提问来源于stack exchange，提问作者Zeruno