求助：用Azure Data Factory以JSON数据集将存储Blob复制到ADX并映射Schema

前置准备 #

• 已创建ADX集群及目标表（若需自动建表，需在ADX数据集配置中指定表结构）
• Blob存储中存储的是结构化JSON文件（非二进制，支持行分隔JSON或JSON数组格式）
• ADF已配置好Blob存储和ADX的服务连接

1. 配置源数据集（Blob JSON） #

• 选择Azure Blob Storage类型，格式选JSON（不要选Binary）
• 设置文件路径，在JSON格式配置中：
  • 若文件是每行一个JSON对象，选「行分隔的JSON」
  • 若文件是JSON数组结构，选「JSON数组」
• 其他默认配置即可

2. 配置目标数据集（ADX） #

• 选择Azure Data Explorer类型，关联已创建的ADX服务连接
• 指定目标数据库和表名：
  • 若目标表已存在，直接填写表名
  • 若需自动创建表，勾选「如果不存在则创建」，并定义表的Schema（字段名+数据类型）

3. 配置复制活动（核心步骤） #

• 源：选择刚才创建的Blob JSON数据集，设置读取范围（如递归读取文件夹、指定文件名通配符）
• 目标：选择ADX数据集
• 映射（Schema映射关键）：
  1. 点击「导入Schema」，可从源数据集或ADX目标表导入字段结构
  2. 手动调整字段映射关系：若源JSON字段与ADX表字段名/类型不一致，逐个匹配（比如源user_id映射到目标UserId，类型对应String）
  3. 嵌套JSON处理：用JSON路径表达式（如$['address']['city']）映射到ADX的扁平字段，或用ADX的dynamic类型存储嵌套对象

复制活动JSON模板（ARM片段） #

{
    "name": "CopyBlobJsonToADX",
    "type": "Copy",
    "policy": {
        "timeout": "7.00:00:00",
        "retry": 0,
        "retryIntervalInSeconds": 30,
        "secureOutput": false,
        "secureInput": false
    },
    "typeProperties": {
        "source": {
            "type": "JsonSource",
            "storeSettings": {
                "type": "AzureBlobStorageReadSettings",
                "recursive": true,
                "wildcardFileName": "*.json"
            },
            "formatSettings": {
                "type": "JsonReadSettings",
                "filePattern": "setOfObjects" // 行分隔JSON用setOfObjects，数组用arrayOfObjects
            }
        },
        "sink": {
            "type": "AzureDataExplorerSink",
            "flushImmediately": false
        },
        "translator": {
            "type": "TabularTranslator",
            "mappings": [
                {
                    "source": {"path": "$['user_id']"},
                    "sink": {"name": "UserId", "type": "String"}
                },
                {
                    "source": {"path": "$['user_name']"},
                    "sink": {"name": "UserName", "type": "String"}
                },
                {
                    "source": {"path": "$['register_time']"},
                    "sink": {"name": "RegisterTime", "type": "DateTime"}
                },
                {
                    "source": {"path": "$['is_active']"},
                    "sink": {"name": "IsActive", "type": "Boolean"}
                },
                {
                    "source": {"path": "$['points']"},
                    "sink": {"name": "Points", "type": "Int32"}
                }
            ]
        }
    },
    "inputs": [{"referenceName": "BlobJsonDataset", "type": "DatasetReference"}],
    "outputs": [{"referenceName": "ADXTargetDataset", "type": "DatasetReference"}]
}

额外注意事项 #

• 复杂场景可在ADX中预先创建ingestion mapping，然后在ADF的sink配置中填写ingestionMappingName，无需在ADF中手动写映射
• 测试阶段先用小文件验证，查看ADX数据导入结果，确认字段类型匹配
• 首次使用建议先通过ADF可视化界面配置，再导出ARM模板参考，更容易理解配置逻辑

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