如何通过Azure Data Factory更新Blob存储中的JSON文件

我需要通过Azure Data Factory完成Blob存储中JSON文件的以下更新操作：

• 根据输入的表名定位JSON中的对应表条目
• 为指定表更新lasttrackingver字段值
• 只要有表条目被更新，就修改根节点的lastupdate字段
• 更新指定表对应的tableupdate字段

示例数据 #

{
  "name" :"job schedule", 
  "lastupdate": "2023-09-15T13:00:00.000000Z", 
  "schedule":{
    "BANKtable" :
    [
      { 
        "tablename":"mani",
        "pipeline":"manikana",
        "tableupdate": "2023-09-15T13:00:00.000000Z",
        "lasttrackingver":"0"
      },
      { 
        "tablename":"ganesh",
        "pipeline":"ganeshkulkarni",
        "tableupdate": "2023-09-15T13:00:00.000000Z",
        "lasttrackingver":"0"
      },
      { 
        "tablename":"jagadesh",
        "pipeline":"jaga",
        "tableupdate": "2023-09-15T13:00:00.000000Z",
        "lasttrackingver":"0"
      }
    ]
  }
}

输入参数 #

• tablename: ganesh
• lasttrackingver: 2

预期输出 #

{
  "name" :"job schedule", 
  "lastupdate": "TODAYSDATE", 
  "schedule":{
    "BANKtable" :
    [
      { 
        "tablename":"mani",
        "pipeline":"manikana",
        "tableupdate": "2023-09-15T13:00:00.000000Z",
        "lasttrackingver":"0"
      },
      { 
        "tablename":"ganesh",
        "pipeline":"ganeshkulkarni",
        "tableupdate": "TODAYSDATE",
        "lasttrackingver":"2"
      },
      { 
        "tablename":"jagadesh",
        "pipeline":"jaga",
        "tableupdate": "2023-09-15T13:00:00.000000Z",
        "lasttrackingver":"0"
      }
    ]
  }
}

实现方案 #

方法1：使用Data Flow（推荐）

1. 源数据集：配置Blob存储中的JSON文件作为源，设置为Single Document模式。
2. 派生列活动：
   • 处理根节点lastupdate：用表达式currentUTC()生成当前时间，直接替换原字段。
   • 处理schedule.BANKtable数组：使用map()函数遍历数组，判断每个条目的tablename是否匹配输入参数：

     map(schedule.BANKtable, 
         if(item().tablename == $tablename, 
            merge(item(), 
                  createMap('lasttrackingver', $lasttrackingver, 'tableupdate', currentUTC())), 
            item()
         )
     )
     

   • 用派生的新数组替换原schedule.BANKtable字段。
3. 接收器数据集：配置回原Blob路径，覆盖原文件。

方法2：使用Lookup + 变量 + Azure Function（适合复杂场景）

1. Lookup活动：读取Blob中的JSON文件，将内容保存到变量originalJson。
2. Set Variable活动：
   • 解析originalJson中的schedule.BANKtable数组，遍历找到匹配tablename的条目，更新lasttrackingver和tableupdate字段。
   • 更新根节点的lastupdate为currentUTC()。
3. Azure Function活动：如果ADF表达式处理数组过于复杂，可将修改后的JSON结构传入Azure Function，由函数完成最终的JSON重组，再写回Blob存储。

注意事项

• 确保输入参数tablename和lasttrackingver已在管道中定义为参数。
• currentUTC()生成的时间格式符合ISO 8601，和示例中的格式一致，若需其他格式可使用formatDateTime(currentUTC(), 'yyyy-MM-ddTHH:mm:ss.fffffffZ')。
• 如果JSON中的表数组所在的键（如BANKtable）不固定，可添加动态参数或使用表达式动态获取键名。

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