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

Posted

技术标签:

【中文标题】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 数组拆分将其拆分为多行。

【讨论】:

以上是关于SSIS 在不使用 SQL 的情况下将多行合并并连接成单行的主要内容,如果未能解决你的问题,请参考以下文章

如何在不使用数据框的情况下将一行分解为多行?

如何在不解压的情况下将 gz 文件合并为 tar.gz?

如何在不采用其中更改的情况下将 git 分支声明为合并?

是否可以在不重新编码的情况下将立体声合并/缩混为单声道(m4a 或 opus)

spark sql 在不使用 where 子句的情况下将所有数据加载到内存中

在不创建存储过程的情况下将 PL/SQL 发送到 Oracle