如何在Azure Data Factory中将Synapse Link for Dataverse的CSV转Parquet并修正列名

前提准备 #

• 确保有权限访问Azure Data Lake Storage (ADLS) 中Synapse Link导出的Tables/路径，以及目标存储路径
• 获取Dataverse表的实体元数据：整理出GUID列名与规范列名的映射关系，可通过两种方式获取：
  • 在Dataverse开发者视图中查看表列属性（“逻辑名称”为GUID格式，“显示名称”为规范名称）
  • 调用Dataverse Web API查询元数据，示例请求：https://<your-environment>.crm.dynamics.com/api/data/v9.2/EntityDefinitions(LogicalName='<table-logical-name>')/Attributes，从中提取映射关系

步骤1：创建ADLS数据源 #

• 新建ADLS Gen2数据源，指向存储CSV文件的Tables/路径，配置CSV格式：
  • 按实际导出设置选择分隔符（通常为逗号）
  • 勾选“第一行作为标题”（CSV首行是GUID列名）

步骤2：整理列名映射规则 #

将GUID与规范列名整理为键值对格式，示例：

{
  "7a5b1c2d-3e4f-5g6h-7i8j-9k0l1m2n3o4p": "客户名称",
  "1a2b3c4d-5e6f-7g8h-9i0j-1k2l3m4n5o6p": "创建日期"
}

可将该映射保存为ADF管道参数，或存储在ADLS的JSON文件中（便于后续维护）

步骤3：用数据流完成转换 #

3.1 配置源数据集 #

选择之前创建的ADLS CSV数据源，设置路径指向Tables/下的目标表文件夹；若需批量处理多表，可使用参数化路径

3.2 派生列替换列名 #

添加派生列活动，使用foldLeft函数结合映射关系替换列名：

• 若映射硬编码在数据流中：

foldLeft($$, (acc, col) => 
  addColumn(acc, 
    iif(containsKey(map('7a5b1c2d-3e4f-5g6h-7i8j-9k0l1m2n3o4p' -> '客户名称', '1a2b3c4d-5e6f-7g8h-9i0j-1k2l3m4n5o6p' -> '创建日期'), col), 
        map('7a5b1c2d-3e4f-5g6h-7i8j-9k0l1m2n3o4p' -> '客户名称', '1a2b3c4d-5e6f-7g8h-9i0j-1k2l3m4n5o6p' -> '创建日期')[col], 
        col), 
    col)
)

• 若映射存储在JSON文件中：先添加查找活动读取JSON文件，再在数据流中引用映射参数，替换上述硬编码的map部分为@pipeline().parameters.columnMapping

3.3 可选：选择目标列 #

添加选择活动，过滤并保留需要的规范列

3.4 配置目标数据集 #

新建ADLS Gen2数据源，设置为Parquet格式，指向Data Lake的目标路径（如Processed/Parquet/），开启自动创建表结构

步骤4：组装并运行管道 #

• 将上述活动（可选的查找、数据流）组装为管道
• 若需批量处理多表：添加Get Metadata活动获取Tables/下的所有文件夹，再用For Each活动遍历每个表，传递表名与对应列映射参数

注意事项 #

• 确保映射中的GUID与CSV列名完全匹配（Dataverse导出的GUID列名通常为小写）
• 批量处理场景建议为每个表维护独立映射文件，或通过ADF的Web活动调用Dataverse API动态生成映射
• Parquet格式建议设置Snappy压缩级别，优化存储与查询性能

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