Azure Data Factory中更新Blob Storage内JSON文件的技术求助

需求说明 #

• 更新Blob Storage中的JSON文件，输入参数为表名和lasttrackingver值
• 匹配指定表名的对象，更新其lasttrackingver字段为传入值
• 只要有表被更新，根节点的lastupdate字段替换为当前UTC日期（格式如YYYY-MM-DDTHH:MM:SS.000000Z）
• 匹配到的表对象的tableupdate字段替换为当前UTC日期

示例数据 #

原始JSON #

{
  "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": "2024-XX-XXTXX:XX:XX.000000Z", 
  "schedule":{
    "BANKtable" :
    [
      { 
        "tablename":"mani",
        "pipeline":"manikana",
        "tableupdate": "2023-09-15T13:00:00.000000Z",
        "lasttrackingver":"0"
      },
      { 
        "tablename":"ganesh",
        "pipeline":"ganeshkulkarni",
        "tableupdate": "2024-XX-XXTXX:XX:XX.000000Z",
        "lasttrackingver":"2"
      },
      { 
        "tablename":"jagadesh",
        "pipeline":"jaga",
        "tableupdate": "2023-09-15T13:00:00.000000Z",
        "lasttrackingver":"0"
      }
    ]
  }
}

具体实现步骤 #

1. 定义管道参数 #

在ADF管道中创建两个字符串类型参数：

• p_tablename: 传入要更新的目标表名
• p_lasttrackingver: 传入要设置的lasttrackingver值

2. 读取原始JSON文件 #

添加Lookup活动，配置如下：

• 数据源连接到目标Blob Storage账户，选择对应的容器和JSON文件
• 取消勾选First row only，确保读取完整的JSON结构

3. 用Data Flow处理JSON转换 #

添加Data Flow活动，将Lookup活动的输出作为输入：

3.1 配置数据源

• 数据源类型选择Inline，格式设为JSON
• 输入内容填写@activity('LookupJSON').output.value（替换Lookup活动的实际名称）

3.2 处理根节点日期

添加派生列转换，创建派生列lastupdate，表达式为：

currentUTC()

该函数生成符合要求的UTC日期格式，若需要本地日期可改用currentDate()并配合toString()调整格式。

3.3 展开数组

添加解析数组转换，选择要展开的数组字段schedule.BANKtable，将数组中的每个对象转换为单独的行。

3.4 更新目标表字段

添加派生列转换，创建两个派生列：

• lasttrackingver：

  iif(tablename == $p_tablename, $p_lasttrackingver, lasttrackingver)
  

• tableupdate：

  iif(tablename == $p_tablename, currentUTC(), tableupdate)
  

3.5 重组数组结构

添加聚合转换：

• 分组依据选择根节点的name、lastupdate以及schedule（保持原结构的非数组字段）
• 聚合字段设置为BANKtable，使用collect()函数将处理后的行重新组合为数组：

  collect(*)
  

3.6 整理最终结构

添加选择转换，确保输出的JSON结构与原始结构完全一致，保留所有必要字段。

4. 写回Blob Storage #

添加Copy活动：

• 数据源选择Data Flow的输出
• 目标连接到原Blob Storage的同一文件路径
• 设置写入行为为Overwrite，确保覆盖原始文件

注意事项 #

• 测试时使用Debug模式逐步骤验证每个活动的输出，确保字段映射和转换逻辑正确
• 若JSON结构有嵌套层级变化，需调整Data Flow中的字段路径
• 确保Blob Storage的权限配置正确，ADF服务主体拥有读写权限

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