基于JSON规则集筛选ADF动态数据集行的技术求助

核心思路 #

通过规则集预处理+动态列引用+批量表达式校验的组合方案，解决设计时未知列的动态校验与打标需求，具体流程围绕「规则缓存→交叉配对→动态校验→过滤打标」展开。

步骤拆解 #

1. 规则集预处理与缓存

• 加载规则：用ADF Lookup活动读取JSON规则集，将规则数组转成行结构（用flatten()函数拆分每组规则）。
• 生成校验表达式：在Lookup后的Derived Column中，为每组规则生成可执行的校验表达式字符串，核心是用byName()动态引用未知列：

  concat(
    '(',
    stringAgg(
      case(
        conditions.operator == 'eq', concat('byName(''', conditions.column, ''') eq ', if(isInteger(conditions.value), conditions.value, concat('''', conditions.value, '''')) ),
        conditions.operator == 'regexMatch', concat('regexMatch(byName(''', conditions.column, '''), ''', replace(conditions.value, '''', ''''''), ''')'),
        conditions.operator == 'gt', concat('byName(''', conditions.column, ''') > ', conditions.value)
      ),
      ' && '
    ),
    ')'
  )
  

  示例：规则Control=1且ScanAddress匹配^192\.168\..*会生成表达式：(byName('Control') eq 1 && regexMatch(byName('ScanAddress'), '^192\.168\..*'))
• 缓存规则：将预处理后的规则写入Data Flow的Cache Sink，供后续流引用。

2. 源数据与规则交叉配对

• 接入动态数据源（设计时未知列），用Cross Join将每行数据与所有规则组配对，确保每行数据能遍历所有规则校验。
• 性能优化：开启Broadcast Join，将小体积的规则集广播到计算节点，降低数据传输开销。

3. 动态规则校验与过滤

• 添加Derived Column，用evaluate()函数执行规则的校验表达式，生成布尔列IsMatch：

  evaluate(ValidationExpr)
  

• 用Filter转换保留IsMatch == true的行，完成不符合规则数据的过滤。

4. 打标与结果输出

• 若一行数据匹配多组规则，用Rank转换按规则优先级（需在规则集增加Priority字段）排序，取Rank=1的行实现唯一打标。
• 用Select转换保留源数据所有列（用*通配符），同时添加规则中的Class、Category标签列。
• 将最终结果写入Delta Lake Sink，配置对应路径与表属性即可。

关键细节处理 #

• 列不存在兼容：若规则中指定的列可能不存在，可在条件表达式前追加isPresent('列名') && ，避免byName()返回null导致校验失败。
• 字符串转义：用replace(conditions.value, '''', '''''')转义规则中的单引号，避免表达式语法错误。
• 数值型列适配：通过isInteger()判断规则值类型，生成表达式时自动去掉数值的引号，保证类型匹配。

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