DataStage中如何对比连续行并按条件批量保留/拒绝行？

核心问题分析 #

你遇到的问题本质是需要成对判断两行数据并批量输出结果，而Transformer默认是逐行处理，没法直接回溯上一行并同时决定两行的去留，所以得通过「缓存上一行+成对校验」或者「分组聚合校验」的方式实现。

方案一：Transformer缓存上一行实现成对输出 #

这个方案适合严格按line1→line2顺序排列的输入数据：

1. 配置Stage Variables #

在Transformer中添加两个用于缓存上一行数据的Stage Variables：

• sv_CachedID：初始值设为NULL，用于存储上一行（line1）的ID
• sv_CachedMessage：初始值设为NULL，用于存储上一行（line1）的MESSAGE

2. 编写输出逻辑 #

在输出链接的列Derivation中，通过Output()函数实现匹配时批量输出两行：

对于ID列：

If Current.MESSAGE Like "%line2" Then
  If sv_CachedID = Current.ID And sv_CachedMessage = Replace(Current.MESSAGE, "line2", "line1") Then
    (Output(YourOutputLinkName, sv_CachedID, sv_CachedMessage); Current.ID)
  Else
    (sv_CachedID := NULL; sv_CachedMessage := NULL; NULL)
Else
  (sv_CachedID := Current.ID; sv_CachedMessage := Current.MESSAGE; NULL)

对于MESSAGE列：

If Current.MESSAGE Like "%line2" Then
  If sv_CachedID = Current.ID And sv_CachedMessage = Replace(Current.MESSAGE, "line2", "line1") Then
    (Output(YourOutputLinkName, sv_CachedID, sv_CachedMessage); Current.MESSAGE)
  Else
    (sv_CachedID := NULL; sv_CachedMessage := NULL; NULL)
Else
  (sv_CachedID := Current.ID; sv_CachedMessage := Current.MESSAGE; NULL)

3. 逻辑说明 #

• 当处理到line1行时，只缓存数据，不输出
• 当处理到line2行时，校验是否和缓存的line1行匹配（ID相同、MESSAGE前缀一致）：
  • 匹配成功：先输出缓存的line1行，再输出当前的line2行
  • 匹配失败：清空缓存，两行都不输出
• 若最后一行是无配对的line1，会因为没有后续的line2行而被自动丢弃

方案二：Aggregator分组校验（更易维护） #

如果输入数据的顺序可能有波动，或者想简化逻辑，用分组聚合的方式更稳妥：

1. 生成分组键 #

先在Transformer中新增一列GroupKey，提取MESSAGE的前缀部分（比如从1-line1中取出1-），Derivation表达式：

Left(Current.MESSAGE, Index(Current.MESSAGE, "-", 1))

2. 分组聚合 #

将数据送入Aggregator Stage，按ID和GroupKey作为分组键：

• 添加聚合列RowCount，用COUNT(*)统计每组的行数
• 添加聚合列MessageList，用LIST(MESSAGE)收集组内所有MESSAGE值

3. 过滤输出 #

再接入一个Transformer，设置约束条件：

RowCount = 2 And Index(MessageList, GroupKey||"line1", 1) > 0 And Index(MessageList, GroupKey||"line2", 1) > 0

满足条件的分组，就输出组内的两行数据（可以通过拆分MessageList实现）。

注意事项 #

• 如果原始输入数据无序，必须先通过Sort Stage按ID和MESSAGE排序，确保line1始终在line2前面
• 方案一中的YourOutputLinkName要替换成你实际的输出链接名称

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