Azure Data Factory中如何仅保留CSV有数据的列？

解决思路 #

依托ADF数据流的架构漂移支持+动态列表达式，先判断列是否存在且包含有效数据，再仅将符合条件的列输出到目标文件。

具体操作步骤 #

1. 保留源的架构漂移配置 #

你的源已经开启了allowSchemaDrift: true和validateSchema: false，这是处理动态列的核心前提，无需修改：

source(output(
          Delivery_Address1 as string,
          Delivery_Address2 as string,
          Delivery_Address3 as string,
          Delivery_Address4 as string,
     ),
     allowSchemaDrift: true,
     validateSchema: false,
     ...) ~> ReadSourceFiles

2. 新增Select转换筛选有效列 #

在TransformFields之后、ConsolidateAndOutputFile之前，插入一个Select转换（命名为FilterValidColumns），通过表达式动态筛选符合条件的列：

• 固定保留的列（如Delivery_Address1/2/3）直接选择
• 可选列（如Delivery_Address4）仅在列存在且非空/非空字符串时保留

对应的数据流脚本片段：

TransformFields select(mapColumn(
          Delivery_Address1,
          Delivery_Address2,
          Delivery_Address3,
          // 条件判断Delivery_Address4：存在且有数据才映射
          case(
              exists(Delivery_Address4) && !isNull(Delivery_Address4) && !isEmpty(Delivery_Address4),
              Delivery_Address4,
              null()
          ) as Delivery_Address4
     ),
     skipDuplicateMapInputs: true,
     skipDuplicateMapOutputs: true) ~> FilterValidColumns

3. 修改Sink为自动映射 #

将Sink的固定列映射改为自动映射，确保仅输出Select转换后保留的列。修改后的Sink脚本：

FilterValidColumns sink(allowSchemaDrift: true,
     validateSchema: false,
     partitionFileNames:[($outputFilename)],
     skipDuplicateMapInputs: true,
     skipDuplicateMapOutputs: true,
     quoteAll: true,
     // 替换为自动映射，无需固定列列表
     mapColumn('auto'),
     partitionBy('hash', 1),
     preCommands: [],
     postCommands: []) ~> ConsolidateAndOutputFile

4. 批量处理未知动态列（可选） #

如果除Delivery_Address4外还有其他未知动态列需要处理，可使用filter结合columns()批量筛选：

TransformFields select(mapColumn(
          // 保留固定列
          Delivery_Address1,
          Delivery_Address2,
          Delivery_Address3,
          // 批量筛选其他列：存在且有数据才保留
          *(filter(columns(), 
              c => exists(c) && !isNull(c) && !isEmpty(c) 
              && c !in ['Delivery_Address1','Delivery_Address2','Delivery_Address3']
          ))
     )) ~> FilterValidColumns

关键函数说明 #

• exists(columnName)：检查当前数据流中是否存在指定列
• isNull(columnName)：判断列值是否为NULL
• isEmpty(columnName)：判断列值是否为空字符串
• columns()：返回当前数据流的所有列名列表
• filter(list, condition)：按指定条件过滤列表元素

修改完成后，输出文件将仅包含存在且有有效数据的列，完全匹配需求。

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