Azure Data Factory中Databricks活动输出JSON转Dataframe方法咨询

1. Azure Data Factory 管道参数配置 #

首先在ADF管道的sample2 Notebook活动中新增一个字符串类型的参数，例如命名为sample1_run_output，参数值填写表达式：

@activity('<你的sample1活动实际名称>').output.runOutput

将sample1的输出结果传递给sample2作为入参。

2. sample2 Notebook 转换代码 #

转换为Pandas DataFrame（和sample1输出格式对齐） #

import json
import pandas as pd

# 读取ADF传入的参数
raw_output = dbutils.widgets.get("sample1_run_output")
# 反序列化为Python字典
result_dict = json.loads(raw_output)
# 提取所有指标数据，兼容sample1输出多个索引行的场景
metrics_list = list(result_dict.values())
# 转为Pandas DataFrame
df = pd.DataFrame(metrics_list)

转换完成的df会自动包含mean、max、min、std、median、L1、MAD、percent_above_mean、variation_from_mean所有需要的字段，单索引输出的场景下df只有1行数据。

转换为Spark DataFrame（适合后续用Spark算子处理的场景） #

import json
from pyspark.sql import Row

# 读取ADF传入的参数
raw_output = dbutils.widgets.get("sample1_run_output")
result_dict = json.loads(raw_output)
metrics_list = [Row(**v) for v in result_dict.values()]
# 转为Spark DataFrame
spark_df = spark.createDataFrame(metrics_list)

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