SSIS 在不使用 SQL 的情况下将多行合并并连接成单行

Posted 2023-03-24

【中文标题】SSIS 在不使用 SQL 的情况下将多行合并并连接成单行

【英文标题】：SSIS consolidate and concatenate multiple rows into single rows without using SQL

【发布时间】：2020-09-16 09:44:54

【问题描述】：

我正在尝试完成一些在 SQL 中很容易完成的事情，但在不使用 SQL 的情况下在 SSIS 中完成似乎非常具有挑战性。基本上，我需要合并和连接一个多对一关系的字段。

给定实体：[合同项目]（许多）到（一个）[帐户]

有一个字段 [ari_productsummary] 包含 Contract Item 实体中列出的产品。我们希望将该值作为 [ari_activecontractitems] 写入 Account。但是，Account 可能有多个与之关联的Contract Item 记录，在这种情况下，我们希望连接这些值。我们也只希望连接不同的值（在我的数据流中已经解决了不同的行）。

这可以通过写入临时表来完成，然后使用查询或视图来获取汇总结果，如下所示。我创建了一个名为 TESTTABLE 的 SQL 表，其中包含来自 Contract Item 实体的 [ari_productsummary] 以及引用的 [accountid em>] 将其映射回 帐户。然后我写了以下查询作为视图：

SELECT distinct accountid,
        (SELECT TT2.ari_productsummary + '; ' 
                FROM TESTTABLE TT2
                WHERE TT2.accountid = TT.accountid
                FOR XML PATH ('')
            ) AS 'ari_activecontractitems'
FROM TESTTABLE TT

执行该查询为我提供了我想要的结果，然后我可以将其用于导入 Account 实体，如下所示：

但是我如何在 SSIS 数据流中做到这一点而不写入 SQL 表作为数据的临时占位符？？我想在里面完成整个过程一个数据流容器，不使用临时 SQL 表/视图。整个总结过程需要即时完成：

是否有人提供不需要临时 SQL 表/视图/查询但完全包含在数据流中的解决方案？

我正在使用 VS 2017 和 KingswaySoft Dynamic CRM 365 ETL 工具集来开发我的解决方案/包。

【问题讨论】：

我没有想到任何原生组件。我认为您不只是在执行 SQL 任务中直接更新帐户的原因（完全跳过临时表和数据流任务是因为它是动态的，而且那里的一切都更难？）

@billinkc 正确，无法在执行 SQL 任务中更新帐户，必须使用 KingswaySoft 工具在数据流任务中读取/写入 Dynamics。我不介意我最初的解决方案，因为我认为它运行良好，但我的老板希望我消除以后可能成为故障点的任何步骤，并认为 SQL 服务器是一个潜在的故障点，并希望我探索所有选项。

【参考方案1】：

在这里吐个球，因为我没有动态，也没有自定义组件。

数据流 1 - 合约聚合

此数据流的目的是在您提供的优雅查询中复制您的逻辑，并将其推送到 Cache Connection Manager（请参阅末尾的 2008+ 注释）

KingswaySoft Dynamics Source -> Script Task -> Cache Transform

如果您想保留排序，请在脚本任务之前执行。我将对 Script Task 采用的实现是它是完全阻塞的——也就是说，所有行都必须到达，然后才能发送任何内容。 Merge Join 之类的任务只是部分阻塞，因为排序数据的要求意味着一旦您不再有当前项目的匹配项，您可以将其发送到管道中。

脚本任务将是asynchronous transformation。您将有两个输出列，您的键 accountid 和新的派生列 ari_activecontractitems。该列可能需要很大——您最了解您的数据，但如果它是 Dynamics 中的 blob 类型（> 4k unicode 或 > 8k ascii 字符），那么您必须将数据类型定义为 DT_TEXT/DT_NTEXT

作为输入，您将从您的来源中选择 accountid 和 ari_productsummary。

代码应该很简单。我们要将accumulate the inbound data 放入字典中。

    //  member variable
    Dictionary<string, List<string>> accumulator;

PreProcess 方法，我们将在其中添加它以初始化我们的变量

    // initialize in PreProcess method
    accumulator = new Dictionary<string, List<string>>();

在 OnBufferRowSent 中（名称近似）

    // simulate the inbound queue
    // row_id would be something like Rows.row_id
    if (!accumulator.ContainsKey(row_id))
    
        // Create an empty dictionary for our list
        accumulator.Add(row_id, new List<string>());
    

    // add it if we don't have it
    if (!accumulator[row_id].Contains(invoice))
    
        accumulator[row_id].Add(invoice);
    

一旦您收到没有更多可用数据的信号，您就开始缓冲输出数据。自动生成的代码将包含所有这些的占位符。

    // This is how we shove data out the pipe
    foreach(var kvp in accumulator)
    
        // approximately thus
        OutputBuffer1.AddRow();
        OutputBuffer1.row_id = kvp.Key;
        OutputBuffer1.ari_productsummary = string.Join("; ",  kvp.Value);

    

【参考方案2】：

我们有一个即将发布的版本，其中包含一个组件，它可以完全满足您的目标，而无需编写自定义代码。该功能目前处于预览阶段，请与我们联系以获取该功能的私人访问权限。您可以在our website找到我们的联系信息。

更新 - 2020 年 6 月 5 日，作为 2020 年第 1 次发布的结果，我们已通过 https://www.kingswaysoft.com/products/ssis-productivity-pack/ 向公众提供这些组件。我们有两个组件可用于此类用途。组合组件将获取输入值并转换为 SSIS 列中的复合值。 Decomposition 组件则相反，它会接受一个输入值并使用基于分隔符的文本拆分或 XML/JSON 数组拆分将其拆分为多行。