65.9K
CodeProject 正在变化。 阅读更多。
Home

一个不可或缺的 SSIS 转换组件 - Script Component

Niladri_Biswas
starIconstarIconstarIconstarIcon
emptyStarIcon
starIcon

4.82/5 (17投票s)

2011 年 5 月 10 日

CPOL

8分钟阅读

viewsIcon

154803

downloadIcon

2606

本文将通过两个实际示例演示如何处理 SSIS Script component。

一个不可或缺的 SSIS 转换组件 - Script Component

目录

  1. 引言
  2. 背景
  3. 示例 1：使用 Script Component 转换在 SSIS 中实现字符串拆分程序。
  4. 示例 2：使用 Script Component 解析 XML 文件，并通过 Bulk Insert Task 将其插入数据库。
  5. 结论

引言

Script component 是一个 SSIS 转换组件，其任务是运行自定义脚本代码。很多时候，我们遇到的情况是没有内置的转换组件；但是，我们可以通过编写一些代码片段来实现所需的转换。这时 Script component 就派上用场了。

背景

转换是大多数 SSIS 生命周期中不可或缺的一部分。一旦原始数据到了我们手中，转换组件的职责就是进行必要的形态变化，并将数据带入所需的格式。尽管 SSIS 中有各种各样的转换可用，但有时我们需要某种自定义转换，而这类组件却不存在。在这种情况下，我们可以使用 Script component，或者创建我们自己的自定义组件。在本文中，我们将通过两个实际示例来演示 Script component 转换的实际应用，而自定义组件的使用将在另一篇文章中讨论。

示例 1：使用 Script Component 转换在 SSIS 中实现字符串拆分程序。

背景

在这个程序中，我们将读取文件内容，如下所示

Id	Value
1	Name1, Name2, Name3
2	Name4, Name5, Name1

并通过 Script Component 转换，我们将得到如下输出

1.jpg

需要执行的步骤

步骤 1

打开 BIDS。从可用的项目类型中选择 Integration Services Project。将 Dataflow Task 拖放到 Control Flow 设计器中。

将 Flat File Source 拖放到 Data Flow 设计器中。右键单击 Flat File Source 组件，然后从弹出菜单中单击 Edit… 来打开 Flat File Source Editor。或者，您可以双击 Flat File Source 组件来弹出 Flat File Source Editor。在 Flat File Source Editor 的连接管理器中，单击 New 按钮并指定源文件。

2.jpg

在 Columns 选项卡中，Row delimiter 应为 {CR}{LF}，而 Columns delimiter 应为 Tab {t}。

3.jpg

单击“确定”按钮。

第二步

添加一个 Script Component Transformation 并将其设置为转换。

4.jpg

步骤 3

从 Flat file Source 到 Script component 添加 precedence constraint。右键单击 Script component，然后从弹出菜单中单击 Edit… 来打开 Script Transformation Editor。

步骤 4：配置脚本组件

在 Input Columns 选项卡中，添加两个可用的列：Column 0，Column 1。

5.jpg

在 Input and Outputs 选项卡中，选择 Output 0，并将其重命名为 Result。将 SynchronousInputID 属性设置为 None，这将使脚本组件变为异步。

6.jpg

注意 ~ SSIS 中有两种类型的转换。

同步转换

输出与输入同步，并且输入数据将逐行处理。

异步转换

输出与输入不同步。所有输入数据将首先被获取，然后读取所有行，最后生成输出。

在 Result 下添加以下输出列：

列名 DataType
ID string [DT_STR]
CustomerName string [DT_STR]

7.jpg

在 Script 选项卡中，将 Script Language 设置为 Microsoft Visual C# 2008，然后单击 Edit Script 按钮。

重写 Input0_ProcessInput 方法以获取直到文件末尾的所有数据，如下所示：

public override void Input0_ProcessInput(Input0Buffer Buffer)
    {
        while (Buffer.NextRow())
        {
            Input0_ProcessInputRow(Buffer);
        }

        if (Buffer.EndOfRowset())
        {
            ResultBuffer.SetEndOfRowset();
        }
    }

接下来，我们需要重写 Input0_ProcessInputRow 方法以将新行添加到输出，如下所示：

public override void Input0_ProcessInputRow(Input0Buffer Row)
    {
      var arr = Row.Column1.Split(','); // Splitting the rows of Names column

      foreach (string str in arr)
      {
         ResultBuffer.AddRow(); //Adding rows to the Result Buffer

         //If the Names are not empty or Null, then set the values to the  
         //  corresponding Result Buffer properties
            if (!string.IsNullOrEmpty(Row.Column1)) 
            {
                ResultBuffer.ID = Row.Column0;
                ResultBuffer.CustomerName = str;
            }
        }
    }

构建应用程序并关闭它。

步骤 5

添加一个 Row Sampling 并启用数据查看器。最终的包设计如下所示：

8.jpg

现在让我们运行包，我们将看到以下输出：

9.jpg

从示例 1 中我们学到了什么

通过这个小程序，我们学会了如何配置组件、编写转换脚本、异步和同步模式是什么等等。现在，我们可以继续进行更复杂的转换，使用这个强大的组件。我们将在下一节中看到一个这样的示例。

示例 2：使用 Script Component 解析 XML 文件，并通过 Bulk Insert Task 将其插入数据库。

背景

考虑以下 XML 格式的玩家信息样本。

<Players>
	<BelongsTo CountryName = "Some Country">
		<Player Id = "1" Name = "Player Name" MatchPlayed="1"  RunsMade = "1" WicketTaken="1"/>	
	</BelongsTo>
</Players>

这是一个简单的 XML 结构，其中包含一些玩家属性，如 ID、姓名等，以及玩家所属国家的信息。

我们有许多这样的玩家信息 XML 文件作为我们的源（为方便实验，我们将只使用四个 XML 文件，名为 PlayerList[n].xml，其中 n=1...4]。源 XML 文件包含在压缩文件中。

总之，我们的任务是读取 XML 文件中的玩家信息，并将它们一次性插入数据库。

需要执行的步骤

步骤 1

打开 BIDS。从可用的项目类型中选择 Integration Services Project。按以下顺序将以下组件拖放到 Control Flow 设计器中：

  1. 1. Execute SQL Task
  2. 1. Dataflow Task 嵌入在 1 Foreach Loop Container 中
  3. 1. Bulk Insert Task。

10.jpg

然后在 Data Flow 设计器中，我们添加一个 Script Component。请注意，在 Select Script Component Type 弹出窗口中，单选按钮应设置为 "Source"。

11.jpg

步骤 2：配置 Execute SQL Task

双击 Execute SQL Task 来打开 Execute SQL Task Editor，并进行以下更改：

12.jpg

在 SQL Statement 部分，我们写下以下查询：

-- Drop the table if it exists
IF EXISTS (SELECT * FROM sys.objects WHERE name = N'tbl_PlayerInfo' AND type = 'U')
    DROP TABLE tbl_PlayerInfo
GO
--Create the table
CREATE TABLE [dbo].[tbl_PlayerInfo](
	[PlayerId] [int] NOT NULL,
	[PlayerName] [varchar](50) NOT NULL,
	[BelongsTo] [varchar](50) NOT nULL,
	[MatchPlayed] [int] NOT NULL,
	[RunsMade] [int] NOT NULL,
	[WicketTaken] [int] NOT NULL
) ON [PRIMARY]
GO

关于 Execute SQL Task 配置就到这里。

步骤 3：配置 Foreach Loop Container

双击 Foreach Loop container 来打开 Foreach Loop Editor。

13.jpg

在 Collection 选项卡中的各种 Foreach enumerators 中，我们只选择 Foreach File enumerator。我们还将指定源文件夹，由于我们只处理 XML 文件，因此 Files 将按 .xml 扩展名过滤，如下所示。

14.jpg

在 variable mapping 部分，我们将选择 <New Variable>。

15.jpg

Add Variable 屏幕将配置如下：

16.jpg

单击 OK。由于每次迭代后，Foreach enumerator 只返回一个值，因此变量的索引应映射到 0。

17.jpg

单击 OK 保存 Foreach Loop container 的设置。

步骤 4：配置 Script Component

双击 Script Component 来打开 Script Transformation Editor。在左侧的 Script 选项卡中，在 ReadOnlyVariable Properties 中设置用户定义的变量 (varCompleteFileName)。为此，让我们首先单击下方图示中标记为 1 的 elipses (...）。对于 Select Variables 屏幕，选择用户定义的变量，在本例中为 varCompleteFileName（在图中标记为 2）。单击 OK 按钮，值将如标记为 3 的图所示进行设置。

18.jpg

接下来，在 Input and Outputs 选项卡中，选择 Output 0，并将其重命名为 PlayerProcessedInfo。在 Output Columns 下添加以下输出列：

列名 数据类型
PlayerID string [DT_STR]
PlayerName string [DT_STR]
BelongsTo string [DT_STR]
MatchPlayed string [DT_STR]
RunsMade string [DT_STR]
WicketTaken string [DT_STR]

此时，看起来如下：

19.jpg

在 Script 选项卡中，将 Script Language 设置为 Microsoft Visual C# 2008，然后单击 Edit Script 按钮。

在 ScriptWindow 中，让我们先创建 XMLDocument Class 和 StreamWriter Class 的对象。

XmlDocument xDoc = new XmlDocument();
StreamWriter sw = null;

在 PreExecute Method 中，让我们像下面一样初始化 StreamWriterClass：

//Initialize the StreamWriter at the PreExecute method.
var destinationFileName = @"D:\All_PlayerInfo.txt";        
sw = new StreamWriter(destinationFileName,true);

正如可以理解的，这甚至可以设置为一个变量并在运行时传递。

接下来，用以下代码片段重写 CreateNewOutputRows()：

//Load the file name
xDoc.Load(Variables.varCompleteFileName);

//Visit the nodes and get the records
 foreach (XmlNode xBelongsToNode in xDoc.SelectNodes("//BelongsTo"))
 {
     foreach (XmlNode xPlayerNode in xDoc.SelectNodes("//Player"))
     {
		var playerInfo = string.Concat(xPlayerNode.Attributes[0].Value,// Player ID
									   ",",
									   xPlayerNode.Attributes[1].Value,// Player Name
									   ",",
									   xBelongsToNode.Attributes[0].Value, // Belongs To
									   ",",
									   xPlayerNode.Attributes[2].Value, //Match Played
									   ",",
									   xPlayerNode.Attributes[3].Value, // Runs Made
									   ",",
									   xPlayerNode.Attributes[4].Value // Wickets taken
									   );

		sw.Write(playerInfo + Environment.NewLine);

	}
}

基本上，我们通过 PlayerNode 循环，以获取每个 XML 文件（源目录中存在）的 PlayerInformation，然后使用 StreamWriter class，我们将信息写入目标位置。

最后，在 PostExecute Method 中，我们释放 StreamWriter 对象。

完整的代码片段如下：

using System;
using System.IO;
using System.Xml;

[Microsoft.SqlServer.Dts.Pipeline.SSISScriptComponentEntryPointAttribute]
public class ScriptMain : UserComponent
{    
    XmlDocument xDoc = new XmlDocument();
    StreamWriter sw = null;   

    public override void PreExecute()
    {
        base.PreExecute();

        //Initialize the StreamWriter at the PreExecute method.
        var destinationFileName = @"D:\All_PlayerInfo.txt";        
        sw = new StreamWriter(destinationFileName,true);        
    }

    public override void PostExecute()
    {
        base.PostExecute();
        //Close the StreamWriter when the work is over
        sw.Close();
    }

    public override void CreateNewOutputRows()
    {    
        //Load the vaious file names
        xDoc.Load(Variables.varCompleteFileName);

		//Visit the nodes and get the records
        foreach (XmlNode xBelongsToNode in xDoc.SelectNodes("//BelongsTo"))
        {
            foreach (XmlNode xPlayerNode in xDoc.SelectNodes("//Player"))
            {
                var playerInfo = string.Concat(xPlayerNode.Attributes[0].Value,// Player ID
                                               ",",
                                               xPlayerNode.Attributes[1].Value,// Player Name
                                               ",",
                                               xBelongsToNode.Attributes[0].Value, // Belongs To
                                               ",",
                                               xPlayerNode.Attributes[2].Value, //Match Played
                                               ",",
                                               xPlayerNode.Attributes[3].Value, // Runs Made
                                               ",",
                                               xPlayerNode.Attributes[4].Value // Wickets taken
                                               );

                sw.Write(playerInfo + Environment.NewLine);

            }
        }
    }
}

如果我们运行应用程序，我们可以看到在 All_PlayerInfo.txt 中生成了以下输出：

1,India Player1,India,10,900,24
2,India Player2,India,12,456,10
3,India Player3,India,6,88,34
4,India Player4,India,9,345,22
5,India Player5,India,11,889,4
1,Australia Player1,Australia,6,500,54
2,Australia Player2,Australia,7,123,5
3,Australia Player3,Australia,15,888,14
4,Australia Player4,Australia,77,5000,223
5,Australia Player5,Australia,55,5567,18
1,Canada Player1,Canada,16,112,64
2,Canada Player2,Canada,5,173,51
1,USA Player1,USA,1,16,3
2,USA Player2,USA,2,14,1

我们的最后任务是使用 Bulk Insert task 将这些记录插入数据库。

步骤 5：配置 Bulk Inset Task

双击 Bulk Insert Task 来打开 Bulk Insert Task Editor。在 Connection 部分，我们进行以下更改：

  1. 提供正确的 Connection String 和 Destination Table Name。
  2. 在此情况下，Column Delimiter 应为 Comma Seperated {,}。
  3. 在此处输入源文件名，即 All_PlayerInfo.txt。

20.jpg

就是这样。

步骤 6

现在让我们运行包，它将如以下所示顺利执行：

21.jpg

为了验证我们的玩家记录是否已正确插入数据库，让我们在 SSMS 查询窗口中执行以下查询：

	select * from [dbo].[tbl_PlayerInfo]

结果窗格将显示以下结果：

22.jpg

从示例 2 中我们学到了什么

在本课程中，我们学习了如何配置和使用 Execute SQL task、Bulk Insert Task，如何在 Foreach Loop Container 的作用域内使用变量在运行时选择文件名，并在脚本组件中读取它们。

结论

因此，我们已经看到了两个关于脚本组件的小示例，我们学习了如何配置此组件，如何与其他组件结合使用它等等。但这些只是该组件强大功能发挥作用的几种场景。我很快会在这个帖子里添加更多关于这个组件的示例。

感谢阅读。

© . All rights reserved.