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 的情况下将多行合并并连接成单行的主要内容,如果未能解决你的问题,请参考以下文章
是否可以在不重新编码的情况下将立体声合并/缩混为单声道(m4a 或 opus)