如何在IBM InfoSphere DataStage中按指定ID聚合字符串并统计出勤

整体处理流程 #

核心流程：读取CSV源文件 → 过滤有效出勤记录 → 按ID排序 → 聚合生成工作日列表与出勤天数 → 输出结果

步骤1：读取源数据 #

用Sequential File组件读取逗号分隔的源文件：

• 配置文件路径，设置分隔符为逗号
• 在「Columns」标签页映射字段：
  • ID：数据类型设为Integer
  • 工作日：数据类型设为VarChar
  • 出勤状态：数据类型设为VarChar

步骤2：过滤出勤记录 #

添加Filter组件，连接到Sequential File的输出端：

• 设置过滤条件：出勤状态 = '是'，只保留实际出勤的记录，减少后续处理量

步骤3：按ID排序 #

添加Sort组件，连接到Filter的输出端：

• 把ID设为排序键（升序），确保相同ID的记录连续排列——这是DataStage分组聚合的必要前提

步骤4：聚合生成目标数据 #

推荐用Transformer组件实现滚动聚合，同时完成工作日拼接和天数统计：

1. 打开Transformer，定义以下阶段变量：

   • sv_Previous_ID：类型Integer，初始值0，用来记录上一条记录的ID
   • sv_Workday_List：类型VarChar，初始值为空字符串，用来累积拼接工作日
   • sv_Attend_Count：类型Integer，初始值0，用来统计出勤天数

2. 在Derivation区域编写字段逻辑：

   • ID字段直接映射输入的ID
   • 工作日列表：

     If sv_Previous_ID = ID Then sv_Workday_List : ',' : 工作日 Else 工作日
     

   • 出勤天数：

     If sv_Previous_ID = ID Then sv_Attend_Count + 1 Else 1
     

3. 更新阶段变量的Derivation：

   • sv_Previous_ID：ID（每次记录当前ID）
   • sv_Workday_List：工作日列表（同步更新拼接后的字符串）
   • sv_Attend_Count：出勤天数（同步更新计数）

4. 给Transformer的输出链接加过滤条件：
   设置LastRow() Or ID <> sv_Previous_ID，只输出每个ID分组的最后一条记录——这条记录包含完整的工作日列表和最终出勤天数

步骤5：输出结果 #

用Sequential File或Database Connector组件连接到Transformer输出端，配置目标文件/表的字段映射，输出最终聚合结果

备选方案：自定义聚合函数（UDA） #

如果需要复用字符串拼接逻辑，可以创建自定义聚合函数：

1. 在DataStage Manager中用C或Basic编写实现字符串拼接的UDA
2. 在Aggregator组件中，用自定义UDA聚合工作日字段，同时用Count()函数统计出勤天数
3. 该方案适合多次复用聚合逻辑的场景，但开发成本略高

内容的提问来源于stack exchange，提问作者Sahan Aloka Mendis