Azure Data Factory动态列透视执行停滞问题排查

问题背景 #

现有多条MATNR相同但其他列内容不同的数据行，需要合并为单一行并整合多行列内容。此前采用以下步骤处理：

• 执行Window转换：以MATNR和CHARG为分区键，通过表达式concat(CHARG, toString(rowNumber()))生成ws列；
• 执行Pivot转换。

单MATNR处理结果正常，但处理大量数据时，ADF生成约65k列（基于25k行），任务运行2小时后自动取消，需排查问题并寻找更优方案。

问题原因 #

原方案的核心问题是Pivot转换生成了远超ADF处理能力的列数：每一行数据都对应生成新列，25k行衍生出65k列，直接超出了ADF Data Flow的内存和处理阈值，并非配置错误或循环问题。

更优实现方案 #

方案1：将多行数据整合成结构化嵌套字段（推荐） #

放弃生成大量列，改为将同MATNR的所有行数据打包为JSON数组或结构化映射，既满足整合需求，又避免列数爆炸：

1. 使用ADF的Aggregate转换，分组键选择MATNR；
2. 对需要整合的字段（LABST、LGORT等）使用collect()函数，生成结构化数组：
   • 示例表达式（生成结构化映射数组）：

     createArray(createMap(
         'LABST', LABST,
         'LGORT', LGORT,
         'LGPBE', LGPBE,
         'CHARG', CHARG,
         'CLABS', CLABS
     ))
     

   • 最终每个MATNR对应一行，整合后的字段为包含所有行数据的数组，后续可按需解析。

方案2：先按CHARG分组合并，再做Pivot #

减少Pivot的列基数：

1. 先执行Aggregate转换，以MATNR+CHARG为分组键，将同一CHARG下的多行字段合并（比如用concat_ws(';', collect(LGORT))合并LGORT值）；
2. 对合并后的结果执行Pivot，此时Pivot的列数等于不同CHARG的数量，远低于原方案的列数。

方案3：用Databricks Notebook替代ADF Data Flow #

针对超大量数据的行列转换，Spark引擎的处理能力远优于ADF Data Flow：

• 在ADF中调用Databricks Notebook，编写Spark SQL或Scala代码实现合并逻辑，比如：

  SELECT 
      MATNR,
      collect_list(struct(LABST, LGORT, LGPBE, CHARG, CLABS)) AS merged_data
  FROM your_table
  GROUP BY MATNR
  

测试用原始数据 #

MATNR;LABST;LGORT;LGPBE;CHARG;CLABS
8200451459;0.0;SU60;101_65-413;Supermarkt;
8200451459;0.0;SU50;101_55-413;Supermarkt;
8200451459;0.0;B417;O 34 01 01;ZN;
8200451459;0.0;B405; ;ZN;
8200451459;0.0;SU15; ;Supermarkt;
8200451459;0.0;B417;O 34 01 01;ZI;
8200451459;0.0;SU30;116_35-413;Supermarkt;
8200451459;0.0;UBWL; ;ZN;
8200451459;515.0;581;O 34 01 01;ZN;163
8200451459;515.0;581;O 34 01 01;ZG;352
8200451459;0.0;SU40;116_45-413;Supermarkt;
8200451459;0.0;586; ;ZD;

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