如何在Azure Synapse笔记本中通过Python SDK获取管道运行ID？

完全可以通过Python代码直接获取当前触发Notebook的Pipeline运行ID，以下是两种原生、无额外依赖的实现方式：

方法1：使用Synapse原生工具mssparkutils（推荐） #

mssparkutils是Synapse内置的工具集，专门用于处理环境变量、文件操作等场景，直接调用getPipelineRunId()就能拿到运行ID：

import pandas as pd

# 获取Pipeline运行ID（仅当Notebook被Pipeline调用时返回有效值）
pipeline_run_id = mssparkutils.env.getPipelineRunId()

# 处理手动运行场景（可选）
if not pipeline_run_id:
    pipeline_run_id = f"manual_run_{pd.Timestamp.now().strftime('%Y%m%d%H%M%S')}"

# 读取并预处理Parquet文件（你的原有逻辑）
processed_df = pd.read_parquet("abfss://<your-container>@<your-storage-account>.dfs.core.windows.net/<input-path>")

# 将运行ID添加到DataFrame
processed_df['pipeline_run_id'] = pipeline_run_id

# 保存到目标Blob
processed_df.to_parquet("abfss://<your-container>@<your-storage-account>.dfs.core.windows.net/<output-path>")

方法2：读取Spark配置参数 #

当Notebook被Pipeline活动调用时，Synapse会自动将运行ID注入到Spark配置的spark.synapse.pipelinerunid项中，你可以通过SparkSession读取这个参数：

from pyspark.sql import SparkSession
import pandas as pd

spark = SparkSession.builder.getOrCreate()
# 获取运行ID，默认值设为空字符串
pipeline_run_id = spark.conf.get("spark.synapse.pipelinerunid", "")

# 后续逻辑同方法1：处理手动运行、添加到DataFrame、保存

注意事项 #

• 两种方法仅在Notebook被Pipeline活动触发运行时会返回真实的Pipeline运行ID；如果是手动打开Notebook执行，会返回空字符串，建议添加手动运行的 fallback 逻辑。
• 获取到的运行ID格式和系统变量@pipeline().RunId返回的完全一致，可直接用于关联Pipeline运行日志或数据溯源。

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