Azure ADF如何仅导出用户列表中变更列及用户ID至JSON

针对每日用户CSV的新增/列变更场景，提供两种实用实现方案：

方案1：使用Power Query（Excel/Power BI环境） #

适用于可视化工具内的操作流程：

1. 导入数据源：分别导入yesterday.csv和today.csv，将UserName设为唯一标识列。
2. 全外连接两张表：以UserName为连接键执行全外连接，覆盖新增、删除、变更的所有行。
3. 筛选目标行：
   • 新增自定义列IsNew，公式：Table.IsNull([yesterday.csv.UserName])（标记今日新增用户）
   • 新增自定义列HasChanges，公式：not Table.IsNull([yesterday.csv.UserName]) and not Table.IsNull([today.csv.UserName]) and ( [yesterday.csv.FirstName] <> [today.csv.FirstName] or [yesterday.csv.LastName] <> [today.csv.LastName] or [yesterday.csv.Department] <> [today.csv.Department] )（标记列值有变更的用户）
   • 筛选出IsNew = true或HasChanges = true的行。
4. 提取变更列：
   新增自定义列ChangedFields，用以下公式生成仅含UserName和变更列的记录：

   let
       currentUser = [today.csv.UserName],
       yesterdayRecord = Record.SelectFields([yesterday.csv], {"FirstName", "LastName", "Department"}),
       todayRecord = Record.SelectFields([today.csv], {"FirstName", "LastName", "Department"}),
       changes = Record.RemoveFields(Record.TransformFields(todayRecord, 
           List.Transform(Record.FieldNames(todayRecord), 
               each {_, (value) => if value <> Record.Field(yesterdayRecord, _) then value else null}
           )), 
           List.Select(Record.FieldNames(todayRecord), each Record.Field(todayRecord, _) = Record.Field(yesterdayRecord, _))
       )
   in
       Record.AddField(changes, "UserName", currentUser)
   

5. 导出JSON：提取ChangedFields列为列表，使用Json.FromValue转换为JSON格式后导出。

方案2：Python脚本（适合自动化批量处理） #

用pandas实现灵活的批量对比：

1. 读取CSV文件：

   import pandas as pd
   import json
   
   df_yesterday = pd.read_csv('yesterday.csv').set_index('UserName')
   df_today = pd.read_csv('today.csv').set_index('UserName')
   

2. 定位目标用户：

   # 新增用户：今日有、昨日无
   new_users = df_today.index.difference(df_yesterday.index)
   # 变更用户：两日都存在且至少一列值不同
   changed_users = df_today.index.intersection(df_yesterday.index)[df_today.ne(df_yesterday).any(axis=1)]
   # 合并目标用户集合
   target_users = new_users.union(changed_users)
   

3. 提取变更数据：

   result = []
   for user in target_users:
       user_data = {'UserName': user}
       # 新增用户：保留所有列值
       if user in new_users:
           user_data.update(df_today.loc[user].to_dict())
       # 变更用户：仅保留有差异的列
       else:
           diff_cols = df_today.loc[user].ne(df_yesterday.loc[user])
           changed_cols = diff_cols[diff_cols].index.tolist()
           user_data.update(df_today.loc[user, changed_cols].to_dict())
       result.append(user_data)
   

4. 导出JSON：

   with open('changes.json', 'w') as f:
       json.dump(result, f, indent=2)
   

   运行后会生成符合需求的JSON文件，示例中User01的输出为[{"UserName": "User01", "Department": "Sales"}]。

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