SSIS脚本任务中按WorkOrderId和WorkOrderProductId统计唯一行数

一、在同一个脚本任务内实现统计 #

SSIS的Output Buffer采用流式处理机制，生成行后不会在内存中保留完整数据集供PostExecute()读取，直接遍历Buffer的方式不可行。正确做法是在生成行的过程中实时统计：

1. 在脚本类中定义分组统计字典，存储每个WorkOrderId+WorkOrderProductId组合的计数：

   // 字段类型根据实际数据类型调整，字符串类型可改为Tuple<string, string>
   private Dictionary<Tuple<int, int>, int> _groupCounts;
   

2. 在PreExecute()方法中初始化字典：

   public override void PreExecute()
   {
       base.PreExecute();
       _groupCounts = new Dictionary<Tuple<int, int>, int>();
   }
   

3. 在生成行的代码块中，每次添加行后更新统计字典：

   if (conditionMeets)
   {
       WOProductBuffer.AddRow();
       WOProductBuffer.WorkOrderId = workOrderId;
       WOProductBuffer.WorkOrderProductId = workOrderProductId;
       // 其他字段赋值
   
       // 更新统计计数
       var groupKey = Tuple.Create(workOrderId, workOrderProductId);
       if (_groupCounts.ContainsKey(groupKey))
       {
           _groupCounts[groupKey]++;
       }
       else
       {
           _groupCounts[groupKey] = 1;
       }
   }
   

4. 统计结果的后续使用：

   • 若需写入SSIS变量，可在PostExecute()中将字典值赋值给对应变量；
   • 若需生成带统计值的输出行，可在PostExecute()中遍历字典，调用Output Buffer的AddRow()生成统计行。

二、新建脚本组件（转换）实现统计 #

如果需要保留原始输出行并添加统计值，或想单独拆分统计逻辑，可新建**脚本组件（转换类型）**处理上游输出：

1. 组件配置：

   • 选择“转换”类型，将上游的WOProduct输出设置为该组件的输入；
   • 添加输出（按需选择输出带统计值的原始行，或仅输出分组统计结果），包含WorkOrderId、WorkOrderProductId、WopCount等字段。

2. 脚本内实现逻辑：

   • 定义缓存和统计字典：

     private Dictionary<Tuple<int, int>, int> _groupCounts;
     // 缓存每行字段值（不能直接存Buffer对象，因Buffer会被复用）
     private List<(int WorkOrderId, int WorkOrderProductId, /*其他字段*/)> _cachedRows;
     

   • 在PreExecute()中初始化：

     public override void PreExecute()
     {
         base.PreExecute();
         _groupCounts = new Dictionary<Tuple<int, int>, int>();
         _cachedRows = new List<(int, int, /*其他字段类型*/)>();
     }
     

   • 处理输入行，缓存数据并更新统计：

     public override void Input0_ProcessInputRow(Input0Buffer Row)
     {
         // 缓存当前行字段值
         _cachedRows.Add((Row.WorkOrderId, Row.WorkOrderProductId, /*其他字段赋值*/));
     
         // 更新统计计数
         var groupKey = Tuple.Create(Row.WorkOrderId, Row.WorkOrderProductId);
         if (_groupCounts.ContainsKey(groupKey))
         {
             _groupCounts[groupKey]++;
         }
         else
         {
             _groupCounts[groupKey] = 1;
         }
     }
     

   • 在PostExecute()中生成最终输出：

     public override void PostExecute()
     {
         base.PostExecute();
         // 方式1：输出带统计值的原始行
         foreach (var row in _cachedRows)
         {
             Output0Buffer.AddRow();
             Output0Buffer.WorkOrderId = row.WorkOrderId;
             Output0Buffer.WorkOrderProductId = row.WorkOrderProductId;
             // 其他字段赋值
             Output0Buffer.WopCount = _groupCounts[Tuple.Create(row.WorkOrderId, row.WorkOrderProductId)];
         }
     
         // 方式2：仅输出分组统计结果（每个组合一行）
         // foreach (var kvp in _groupCounts)
         // {
         //     Output0Buffer.AddRow();
         //     Output0Buffer.WorkOrderId = kvp.Key.Item1;
         //     Output0Buffer.WorkOrderProductId = kvp.Key.Item2;
         //     Output0Buffer.WopCount = kvp.Value;
         // }
     }
     

关于之前尝试的问题说明 #

直接将WOProductBuffer加入列表无法获取有效数据，是因为SSIS的Buffer是流式复用对象，AddRow()后行数据会被后续处理流程接管，不会在Buffer中保留完整历史行数据，因此必须在生成/处理行的实时阶段完成统计。

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