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

Question

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

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

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

这可以通过写入临时表，然后使用查询或视图来获得汇总结果，如下所示。我创建了一个名为TESTTABLE的SQL表，其中包含来自[[Contract Item实体的[

ari_productsummary

]以及引用的[accountid]，以将其映射回Account] >。然后，我将以下查询作为视图编写：

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表/视图。整个摘要过程需要随时进行：

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

我正在使用VS 2017和KingswaySoft Dynamic CRM 365 ETL工具集来开发我的解决方案/软件包。我正在尝试完成一些很容易在SQL中完成的任务，但是在不使用SQL的情况下在SSIS中完成似乎非常具有挑战性。基本上，我需要合并和连接...

Answer 1

KingswaySoft Dynamics源->脚本任务->缓存转换

如果要保留排序，请在执行脚本任务之前执行。我将通过脚本任务执行的实现是完全阻塞-这是所有行必须先到达才能发送任何内容。诸如合并联接之类的任务仅受[[partially

阻塞，因为排序数据的要求意味着一旦您不再与当前项目匹配，就可以将其发送到管道中。

脚本任务将为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);

    }