在 SSIS 数据流中组合多种记录类型

Question

我可以使用集成服务来提取和转换数据，例如，包含在导入平面文件中单一类型的固定格式记录中的数据。我还可以在一个文件中处理多种类型的固定格式记录，只要我能够将它们发送到不同的目的地。

但是，如果我需要将输入文件中的两条（或更多条）记录关联到要发送到单个目的地的每条记录，该怎么办？我知道这些记录属于一起的唯一方法是它们在输入文件中彼此相邻。数据流将处理一条又一条记录。我不能使用 SSIS 包变量来计算记录，因为包变量只有在数据流完成后才会更新。

这是一个示例输入文件：

HDR00120140501      
DTL001JOAN     0.00 
DTL002     30.00 ABC
DTL001DAVE    11.00 
DTL002     21.85 DEF
DTL001BERT    50.00 
DTL002      0.00 GHI
TRL001              

我需要将 DTL001 和 DTL002 详细信息行中的数据合并为单个输出记录。此外，我需要包含来自 HDR001 标题行的信息（在本例中为日期值）。所以目标表看起来像：

CREATE TABLE TestImport (
  ID int IDENTITY(1,1) NOT NULL,
  ImportDate datetime NULL,
  Name char(4) NULL,
  Amount1 decimal(18, 2) NULL,
  Amount2 decimal(18, 2) NULL,
  Desc char(3) NULL )

六个输入明细记录，加上表头，将在目标表中产生三个记录：

ImportDate Name Amount1 Amount2 Desc
---------- ---- ------- ------- -----
2014-05-01 JOAN 0.00    30.00   ABC
2014-05-01 DAVE 11.00   21.85   DEF
2014-05-01 BERT 50.00   0.00    GHI

这是一个已解决的问题，旨在帮助可能遇到类似问题的任何人。

Answer 1

可以使用Script、Conditional Split 和Merge Join 组件的组合来组合详细记录。

脚本组件添加一个详细记录计数，该计数会随着它遇到的每个 DTL001 记录而增加。这提供了一个详细记录计数列，可用于对输入文件中相邻的 DTL001 和 DTL002 记录进行排序和连接。

平面文件连接管理器和源用于将每个输入记录分成 RecordType、RecordSubType 和 RecordData 列。例如，前三个记录分为以下几部分：

"HDR", "001", "20140501      "
"DTL", "001", "JOAN     0.00 "
"DTL", "002", "     30.00 ABC"

接下来，Derived Header Columns 组件添加 HeaderRecordCount、DetailRecordCount 和 ImportDate 列。这些最初是未填充的，但它们为脚本组件提供了一个存储标题和详细信息计数的位置，以及目标表的每一行所需的标题中的任何信息。该脚本还可以处理具有多个节的文件，其中每个节都有一个新的 HDR001 头记录：

using System;
using System.Data;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using Microsoft.SqlServer.Dts.Runtime.Wrapper;

[Microsoft.SqlServer.Dts.Pipeline.SSISScriptComponentEntryPointAttribute]
public class ScriptMain : UserComponent
{
    static int _HeaderRecordCount = 0;
    static int _DetailRecordCount = 0;
    static DateTime _ImportDate = DateTime.MinValue;
    public override void Input0_ProcessInputRow(Input0Buffer Row)
    {
        if (Row.RecordType == "HDR" && Row.RecordSubType == "001")
        {
            _HeaderRecordCount = _HeaderRecordCount + 1;
            _ImportDate = DateTime.ParseExact(Row.RecordData.Substring(0, 8), "yyyyMMdd", System.Globalization.CultureInfo.InvariantCulture);
        }
        else if (Row.RecordType == "DTL" && Row.RecordSubType == "001")
        {
            _DetailRecordCount = _DetailRecordCount + 1;
        }
        Row.HeaderRecordCount = _HeaderRecordCount;
        Row.DetailRecordCount = _DetailRecordCount;
        Row.ImportDate = _ImportDate;
    }
}

然后，条件拆分组件将 DTL001 和 DTL002 记录发送到不同的路径。拆分条件为：

RecordType == "DTL" && RecordSubType == "001"
RecordType == "DTL" && RecordSubType == "002"

这些路径都在 DetailRecordCount 列上排序，Merge Join 使用该列将它们组合起来。来自 001 路径的所有记录都传递到 Merge Join 输出，来自 002 路径的 RecordData 列也传递到单独的输出列（例如，RecordData002）。

现在两个平面文件记录的内容都可以在单个数据流记录中使用。 Derived Detail Columns 组件用于提取任何需要的信息，并将生成的列发送到目标表。