Azure DataFlow按错误条件分流数据至Blob不同文件夹的实现问题

核心思路 #

先统计整份数据是否存在错误行，将这个全局判断结果附加到每一行数据上，再根据全局标记分流输出，避免Aggregate转换丢失原数据的问题。

分步实现 #

• 步骤1：统计全局错误状态
  在Derived Column转换之后新增一个Aggregate转换，不设置任何分组键，添加聚合列：

  • 列名：HasError
  • 表达式：sumIf(BadData == true(), 1) > 0
    该表达式会返回布尔值：true表示存在错误行，false表示无错误行。

• 步骤2：将全局标记附加到原数据流
  保留Derived Column后的原数据流作为主分支，将Aggregate分支通过Cross Join转换与主分支合并：
  Cross Join会把Aggregate生成的单行全局标记，附加到主分支的每一行数据上，这样所有原始数据行都会带上HasError标记。

• 步骤3：分流输出到对应文件夹
  添加Conditional Split转换，基于HasError列分流：

  • 分支1条件：HasError == true()，输出到Azure Blob的reject文件夹
  • 分支2条件：HasError == false()，输出到Azure Blob的accept文件夹

关键注意事项 #

• Aggregate转换必须不设置分组键，否则会按分组统计，无法得到整份数据的全局错误状态。
• 使用Cross Join不会导致数据膨胀，因为Aggregate分支仅生成一行统计结果。
• 输出到Blob时，直接在输出目标的文件路径中指定reject或accept文件夹即可。

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