如何在Azure Data Factory中对比两个JSON文件并提取新增ID

需求说明 #

通过ADF对比历史、当前两份JSON文件的ID字段，拆分出仅当前文件存在的新增ID，和两份文件都存在的存量ID，分别执行不同的业务活动。

示例JSON文件 #

历史JSON文件

{
  "count": 2,
  "values": [
    {
      "id": "4e10aa02d0b945ae9dcf5cb9ded9a083"
    },
    {
      "id": "cbc414db-4d08-48f2-8fb7-748c5da45ca9"
    }
  ]
}

当前JSON文件

{
  "count": 3,
  "values": [
    {
      "id": "4e10aa02d0b945ae9dcf5cb9ded9a083"
    },
    {
      "id": "cbc414db-4d08-48f2-8fb7-748c5da45ca9"
    },
    {
      "id": "5ea951e3-88d7-40b4-9e3f-d787b94a43c8"
    }
  ]
}

具体实现步骤 #

• 步骤1：读取两份JSON文件
  新建两个查找活动，分别关联历史JSON和当前JSON的数据集，取消勾选「仅返回第一行」配置，确保读取到完整的JSON内容。
• 步骤2：提取两份文件的ID列表
  新建两个数组类型的变量historicalIds和currentIds，通过设置变量活动分别提取ID集合：
  历史ID提取表达式：@activity('读取历史JSON').output.value[0].values[*].id
  当前ID提取表达式：@activity('读取当前JSON').output.value[0].values[*].id
• 步骤3：拆分新增ID与存量ID
  新建两个数组类型的变量newIds和existingIds，使用ADF集合函数计算结果：
  新增ID（仅当前文件存在）：@except(variables('currentIds'), variables('historicalIds'))
  存量ID（两份文件都存在）：@intersection(variables('currentIds'), variables('historicalIds'))
• 步骤4：分支执行不同活动
  根据业务需要，直接使用ForEach活动遍历newIds执行新增类活动，遍历existingIds执行存量类活动即可。

示例运行结果 #

用提供的示例JSON运行后得到的结果如下：

• 新增ID列表：["5ea951e3-88d7-40b4-9e3f-d787b94a43c8"]
• 存量ID列表：["4e10aa02d0b945ae9dcf5cb9ded9a083", "cbc414db-4d08-48f2-8fb7-748c5da45ca9"]

注意事项 #

• 若你的JSON存储在ADLS Gen2、Blob存储等位置，数据集选择JSON格式、编码设置为UTF-8即可正常解析
• 若实际JSON结构与示例有差异，调整表达式中的字段路径匹配你的实际结构即可

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