Azure Data Factory数据流中通过JSON路径实现表格转动态JSON的方法

ADF数据流原生暂不支持直接通过JSON路径字符串动态给层级节点赋值，针对你提到的EAV表转动态JSON的场景，可通过以下两种方案实现： #

方案1：源侧SQL预构建（性能最优，适合路径规则可枚举的场景） #

直接在源转换的查询语句中用数据库自带的JSON函数提前拼接好结构，无需后续ADF转换处理，示例SQL（适配SQL Server）如下：

SELECT 
    (
        SELECT 
            EntityID AS 'user.id',
            MAX(CASE WHEN AttributePath = 'user.firstName' THEN AttributeValue END) AS 'user.firstName',
            MAX(CASE WHEN AttributePath = 'user.lastName' THEN AttributeValue END) AS 'user.lastName',
            MAX(CASE WHEN AttributePath = 'user.address.city' THEN AttributeValue END) AS 'user.address.city',
            MAX(CASE WHEN AttributePath = 'user.address.state' THEN AttributeValue END) AS 'user.address.state'
        FROM EAV t2 
        WHERE t2.EntityID = t1.EntityID
        FOR JSON PATH, WITHOUT_ARRAY_WRAPPER
    ) AS json_content
FROM EAV t1
GROUP BY EntityID
FOR JSON PATH, WITHOUT_ARRAY_WRAPPER

如果路径规则不固定，可搭配动态SQL生成查询逻辑，无需手动枚举所有路径 #

方案2：数据流+Python UDF（完全动态，适配任意路径结构） #

如果需要完全不依赖源侧SQL，仅通过ADF数据流实现，可按以下步骤操作：

• 第一步：添加聚合转换，按EntityID字段分组，用collect()函数将同实体的路径、值聚合为数组，输出结构为 (EntityID int, attr_list array<(AttributePath string, AttributeValue string)>)
• 第二步：添加「Azure Function转换」或「派生列+自定义表达式」，调用Python函数解析路径数组动态构建JSON，参考代码如下：

def build_dynamic_json(entity_id, attr_list):
    res = {"user": {"id": entity_id}}
    for attr in attr_list:
        path_nodes = attr["AttributePath"].split(".")
        # 跳过最外层固定的user节点，遍历剩余层级
        cur_node = res["user"]
        for i, node in enumerate(path_nodes[1:]):
            if i == len(path_nodes) - 2:
                cur_node[node] = attr["AttributeValue"]
            else:
                if node not in cur_node:
                    cur_node[node] = {}
                cur_node = cur_node[node]
    return res

• 第三步：将所有实体的构建结果聚合后输出为JSON文件即可。

注意：你给出的预期JSON存在语法错误，同一个JSON对象下不允许出现重复的user键，实际使用时建议调整为[{"user": {...}}, {"user": {...}}]的数组结构，或用EntityID作为外层的唯一键。

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