如何将Azure Synapse Notebook的SQL查询结果传递至后续管道活动？

核心问题 #

你之前的表达式返回null，是因为默认情况下Notebook不会自动把SQL查询结果传递给管道，必须在Notebook里主动把结果输出给管道，管道才能读取到。

步骤1：修改Notebook，传递查询结果给管道 #

在你的PySpark Notebook里，执行SQL查询后，用mssparkutils.notebook.exit()方法把结果以JSON格式输出，这样管道就能拿到数据。示例代码如下：

# 执行关联查询
df = spark.sql("SELECT A.name FROM A LEFT JOIN B ON A.id = B.id")
# 收集结果（结果少于50行，直接收集无性能问题）
result_rows = df.collect()
# 把Row对象转成字典再转成JSON字符串
import json
result_json = json.dumps([row.asDict() for row in result_rows])
# 将结果传递给管道
mssparkutils.notebook.exit(result_json)

步骤2：在If Condition活动中访问结果 #

现在管道可以通过@activity('Notebook1').output获取到Notebook返回的JSON字符串，你可以根据需求解析后使用：

• 判断结果集行数是否大于0：

  @greater(length(json(activity('Notebook1').output)), 0)
  

• 获取第一条结果的name字段：

  @json(activity('Notebook1').output)[0].name
  

• 遍历所有结果（结合ForEach活动使用）：
  直接把@json(activity('Notebook1').output)作为ForEach的输入数据集即可。

为什么之前的表达式无效？ #

@activity('Notebook1').output.status.Output.result.exitValue这个路径对应的是Notebook的执行状态元数据，只有当你通过mssparkutils.notebook.exit()输出内容时，exitValue才会被填充为你输出的内容。之前你的Notebook没有主动输出任何内容给管道，所以这个值为null。

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