如何在Azure Data Factory中扁平化JSON并解决PostgreSQL导入空值问题

我需要将API源的数据导入PostgreSQL表。已配置Copy Data活动，但Groups和Questions字段出现空值，请问我的映射配置中缺失了什么？Checklists可包含一个或多个Groups，每个Groups可包含一个或多个Questions。是否有更优的方式扁平化JSON字符串后再导入表？

API返回的JSON结构示例：

{
"checklists": [
    {
        "id": "71Pbxb42",
        "name": "Emergency Planning Checklist",
        "checklist_template_id": "2QyMrEy2",
        "modified_date": "2022-12-12T03:31:14Z",
        "created_date": "2022-12-12T03:31:14Z",
        "total": 13.00,
        "max_total": 13.00,
        "franchisee": {
            "id": "GYNnZEAW",
            "name": "Kindergartens - Huntly",
            "external_id": "CNIKA_5179"
        },
        "groups": [
            {
                "id": "WoNgLAQG",
                "checklist_template_group_id": "73evDRBW",
                "name": "Early childhood services and Mātauranga Ake ",
                "attachments": [],
                "group_score": 13.00,
                "max_score": 13.00,
                "questions": [
                    {
                        "id": "GzRlJk4G",
                        "question_type": "select",
                        "question": "Team roles and responsibilities are clarified in the Emergency Procedure ",
                        "select": "Yes",
                        "score": 1.00,
                        "max_score": 1.00,
                        "comment": "",
                        "checklist_template_question_id": "G66bDbmG"
                    },
                    {
                        "id": "LzRlJm4G",
                        "question_type": "select",
                        "question": "Team roles and responsibilities are clarified in the Emergency Procedure ",
                        "select": "Yes",
                        "score": 1.00,
                        "max_score": 1.00,
                        "comment": "",
                        "checklist_template_question_id": "G66bDbmG"
                    }
                ]
            }
        ]
    }
]
}

一、Groups/Questions字段空值的原因及映射修复 #

空值核心原因：Copy Data活动默认仅提取JSON顶层字段，不会自动解析嵌套数组结构，你的映射配置缺失了对嵌套数组的展开和路径指定：

• 未针对groups数组配置**Flatten（展开）**转换，系统无法识别数组内的字段
• 未处理groups下二级嵌套的questions数组，同样需要指定深层字段路径或二次展开

如果要在Copy Data活动中直接映射，需：

1. 在源数据集的投影设置中，手动指定嵌套字段路径，比如groups[*].id、groups[*].questions[*].question
2. 启用Flatten转换，将数组层级展开为平级字段（但多层嵌套下会产生冗余数据，不推荐）

二、更优的扁平化方案：拆分关系型表 #

由于JSON是多层嵌套结构，最符合PostgreSQL设计的方式是拆分为关联关系型表，避免数据冗余，同时方便后续查询：

1. 表结构设计 #

• checklists（主表）：存储顶层检查清单信息
  字段：id(主键), name, checklist_template_id, modified_date, created_date, total, max_total, franchisee_id, franchisee_name, franchisee_external_id
• checklist_groups（关联表）：存储检查清单分组信息，外键关联checklists.id
  字段：id(主键), checklist_id(外键), checklist_template_group_id, name, group_score, max_score
• group_questions（关联表）：存储分组下的问题信息，外键关联checklist_groups.id
  字段：id(主键), group_id(外键), question_type, question, select_answer, score, max_score, comment, checklist_template_question_id

2. 实现方式 #

方式一：用Data Factory流程处理

• 第一步：Copy Data活动提取顶层checklist数据导入checklists表，保留id作为关联键
• 第二步：Lookup活动获取所有checklist数据，用ForEach活动遍历每个checklist
• 第三步：在ForEach内，对当前checklist的groups数组做Flatten转换，关联checklist.id后导入checklist_groups表
• 第四步：同样在ForEach内，对每个group的questions数组做二次Flatten，关联group.id后导入group_questions表

方式二：用PostgreSQL JSON函数直接导入

如果已将完整JSON导入到临时表（比如temp_json_table，含raw_data jsonb字段），可使用jsonb_array_elements函数展开嵌套数组，执行以下SQL批量导入：

-- 导入checklists主表
INSERT INTO checklists (id, name, checklist_template_id, modified_date, created_date, total, max_total, franchisee_id, franchisee_name, franchisee_external_id)
SELECT 
    (checklist->>'id')::varchar,
    checklist->>'name',
    checklist->>'checklist_template_id',
    (checklist->>'modified_date')::timestamptz,
    (checklist->>'created_date')::timestamptz,
    (checklist->>'total')::numeric,
    (checklist->>'max_total')::numeric,
    (checklist->'franchisee'->>'id')::varchar,
    checklist->'franchisee'->>'name',
    checklist->'franchisee'->>'external_id'
FROM (SELECT jsonb_array_elements(raw_data->'checklists') AS checklist FROM temp_json_table) AS checklists;

-- 导入checklist_groups表
INSERT INTO checklist_groups (id, checklist_id, checklist_template_group_id, name, group_score, max_score)
SELECT 
    (group_data->>'id')::varchar,
    (checklist->>'id')::varchar,
    (group_data->>'checklist_template_group_id')::varchar,
    group_data->>'name',
    (group_data->>'group_score')::numeric,
    (group_data->>'max_score')::numeric
FROM (
    SELECT 
        checklist,
        jsonb_array_elements(checklist->'groups') AS group_data
    FROM (SELECT jsonb_array_elements(raw_data->'checklists') AS checklist FROM temp_json_table) AS checklists
) AS groups;

-- 导入group_questions表
INSERT INTO group_questions (id, group_id, question_type, question, select_answer, score, max_score, comment, checklist_template_question_id)
SELECT 
    (question_data->>'id')::varchar,
    (group_data->>'id')::varchar,
    question_data->>'question_type',
    question_data->>'question',
    question_data->>'select',
    (question_data->>'score')::numeric,
    (question_data->>'max_score')::numeric,
    question_data->>'comment',
    question_data->>'checklist_template_question_id'
FROM (
    SELECT 
        group_data,
        jsonb_array_elements(group_data->'questions') AS question_data
    FROM (
        SELECT 
            checklist,
            jsonb_array_elements(checklist->'groups') AS group_data
        FROM (SELECT jsonb_array_elements(raw_data->'checklists') AS checklist FROM temp_json_table) AS checklists
    ) AS groups
) AS questions;

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