Minimized costs, increased effectiveness, higher flexibility and easy management have become keys to success in the IT industry today. This page aims at achieving the above by introducing a new approach to solve a set of problems faced in ETL process.
The IT industry is ever expanding, and elements of IT can be found in almost every sector, every company. Whether you are playing on a website like (www.casino.com/) or using advanced computerized utilities on a farm, knowledge of IT and programming is going to be very useful. Computer skills are necessary all over the world these days, from basic knowledge to complex procedures. Processing a set of operations with different source and target names, but same structure and same set of transformations is a common scenario in development areas. So approaching an automated methodology for the same using some programming languages like C# .Net or VB .Net with little or no modifications is better. This can reduce the total work load up to 80 – 90%.
This page deals with the approach of creating the ETL Package by using C# .Net and VB .Net language. As SQL Server 2005 is a product of Microsoft Corporation, we got the whole set of APIs in Microsoft Visual Studio. Using these APIs and some standard procedures and scripts in T-SQL, we can ease the ETL Process to increase the productivity.
SQL Server Integration Services (SSIS) has an architecture that separates task management (control flow) and data movement and transformation (data flow). Programming Integration Services can be used very efficiently and effectively to extend and automate the purpose of SSIS i.e. the ETL process. We have two engines for both types of operations.
- Run-Time Engine: The run-time engine implements the control flow and package management infrastructure i.e. for logging, precedence constraints, package configuration, event handlers, and variables. By programming the run-time engine, developers can automate the creation, configuration, logging, precedence constraints and execution of packages and create custom tasks and other extensions.
- Data Flow Engine: The data flow engine is specialized for extracting, transforming, and loading data. The data flow task contains additional objects called data flow components like Data Flow Source components, Data Flow Transformation components and Data Flow Destination components. Programming can lead to automation in the creation of Data Flow components and as well creation of custom components.
Both the run-time engine and the data flow engine are written in native code. As a result they are available through a fully managed object model.
Commonly used Assemblies for programming Integration Services
|
Assembly
|
Description
|
|
Microsoft.SqlServer.ManagedDTS.dll
|
Contains the managed run-time engine.
|
|
Microsoft.SqlServer.RuntimeWrapper.dll
|
Contains the primary interop assembly (PIA), or wrapper, for the native run-time engine.
|
|
Microsoft.SqlServer.PipelineHost.dll
|
Contains the managed data flow engine.
|
|
Microsoft.SqlServer.PipelineWrapper.dll
|
Contains the primary interop assembly (PIA), or wrapper, for the native data flow engine.
|
(http://msdn.microsoft.com/en-us/library/ms403344.aspx)
Below are the basic steps for Programming Integration Services:
Below you will see different approaches taking care of different tasks and transformations commonly used in building SSIS package.
EXAMPLE:
Here in this example, the data from source database named OLTP will be loaded to the destination database OLAP. In the OLTP database, the table named Employee_Dim will be transferred to the OLAP database. And during this transformation, the package log will be tracked in PackageTransaction table in the OLAP database.
Details about PackageTransaction table:
|
Column_name
|
Type
|
|
TransactionID
|
int
|
|
PackageName
|
varchar
|
|
RunDate
|
datetime
|
|
RowCount
|
int
|
|
Status
|
bit
|
Details about Employee_Dim table:
|
Column_name
|
Type
|
|
employee_Sl_No
|
int
|
|
emp_name
|
varchar
|
|
emp_id
|
int
|
|
emp_DOB
|
datetime
|
We have the below data in Employee_Dim table:
|
employee_Sl_No
|
emp_name
|
emp_id
|
emp_DOB
|
|
1
|
Arun
|
101
|
1984-03-16 00:00:00.000
|
|
2
|
Akash
|
102
|
1982-01-31 00:00:00.000
|
|
3
|
Ram
|
103
|
1986-03-03 00:00:00.000
|
|
4
|
Rasi
|
104
|
1986-04-25 00:00:00.000
|
The package structure will be like this:
The variables used for the package are:
Before the loading of data starts, this particular run of the package in PackageTransaction table will be logged. Here the column named TransactionID in PackageTransaction table is an IDENTITY Column, so its value will be generated automatically.
The procedure used to do this is:
CREATE PROCEDURE InitTransaction
AS
BEGIN
insert into dbo.PackageTransaction
( PackageName,
RunDate,
Status
)
values
( ’sample package’,
getdate(),
0
)
SELECT
CAST(Scope_Identity() AS INT) TransactionID
END
GO
The above procedure will return the Transaction ID to the package, so that it can be used while ending the package execution. The Status column indicates the successful completion of the package, 0 means not completed and 1 means successfully completed.
Similarly, we need another procedure to log the completion of the package in PackageTransaction table.
CREATE PROCEDURE [dbo].[EndTransaction]
@TransactionID [int],
@RowCount [int]
AS
BEGIN
UPDATE dbo.PackageTransaction
SET
[RowCount] = @RowCount,
Status = 1
WHERE TransactionID = @TransactionID
END
After executing the package, we will get the below result in the PackageTransaction table.
|
TransactionID
|
PackageName
|
RunDate
|
RowCount
|
|
1
|
sample package
|
2009-05-01 08:55:52.467
|
4
|
Let’s do the same by Integration programming. For this we have to create a package first, and then have to add the tasks and the precedence constraints accordingly, and then have to add the package variable. In the dataflow task, we have to add the data flow source and data flow destination components.
The code below shows the above package programmatically and the code itself is self-descriptive. The code for SQL server 2005 and SQL Server 2008 are a bit different; both the approaches are given below.
Add the below References before developing the code.
Microsoft.SqlServer.Dts.Design
Microsoft.SqlServer.DTSPipelineWrap
Microsoft.SQLServer.DTSRuntimeWrap
Microsoft.SQLServer.ManagedDTS
Programming Integration Service in SQL Server 2005
C# .Net Expand
using System;
using System.Collections.Generic;
using System.Text;
using Microsoft.SqlServer.Dts.Runtime;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using System.Data.SqlClient;
using System.Xml;
//The below code will create a SSIS package for SQL Server 2005
namespace CreatePackageSample
{
class Program
{
static void Main(string[] args)
{
//To Create a package named [Sample Package]
Package package = new Package();
package.Name = "Sample Package";
package.PackageType = DTSPackageType.DTSDesigner90;
package.VersionBuild = 1;
//To add Connection Manager to the package
//For source database (OLTP)
ConnectionManager OLTP = package.Connections.Add("OLEDB");
OLTP.ConnectionString = "Data Source=.;Initial Catalog=OLTP;Provider=SQLNCLI.1;Integrated Security=SSPI;Auto Translate=False;";
OLTP.Name = "LocalHost.OLTP";
//For destination database (OLAP)
ConnectionManager OLAP = package.Connections.Add("OLEDB");
OLAP.ConnectionString = "Data Source=.;Initial Catalog=OLAP;Provider=SQLNCLI.1;Integrated Security=SSPI;Auto Translate=False;";
OLAP.Name = "LocalHost.OLAP";
//To add package variables
Variable TransactionID = package.Variables.Add("TransactionID", false, "User", 0);
TransactionID.Name = @"TransactionID";
Variable RowCountVar = package.Variables.Add("RowCountVar", false, "User", 0);
RowCountVar.Name = @"RowCountVar";
//To add Package Start Indicator (Execute Sql task )
TaskHost PackageStartIndicatorTask = (TaskHost)package.Executables.Add("Microsoft.SqlServer.Dts.Tasks.ExecuteSQLTask.ExecuteSQLTask, Microsoft.SqlServer.SQLTask, Version=9.0.242.0, Culture=neutral, PublicKeyToken=89845dcd8080cc91");
PackageStartIndicatorTask.Name = @"Package Start Indicator";
PackageStartIndicatorTask.Description = @"Execute SQL Task Description";
XmlDocument PackageStartIndicatorTaskdoc = new XmlDocument();
PackageStartIndicatorTaskdoc.LoadXml(@"");
((IDTSComponentPersist)PackageStartIndicatorTask.InnerObject).LoadFromXML(PackageStartIndicatorTaskdoc.DocumentElement, null);
//To add Load Employee Dim to the package [Data Flow Task]
TaskHost dataFlowTaskHost = (TaskHost)package.Executables.Add("DTS.Pipeline.1");
dataFlowTaskHost.Name = @"Load Employee Dim";
dataFlowTaskHost.FailPackageOnFailure = true;
dataFlowTaskHost.FailParentOnFailure = true;
dataFlowTaskHost.DelayValidation = false;
dataFlowTaskHost.Description = @"Data Flow Task";
//-----------Data Flow Inner component starts----------------
MainPipe dataFlowTask = dataFlowTaskHost.InnerObject as MainPipe;
// Source OLE DB connection manager to the package.
ConnectionManager SconMgr = package.Connections["LocalHost.OLTP"];
// Destination OLE DB connection manager to the package.
ConnectionManager DconMgr = package.Connections["LocalHost.OLAP"];
// Create and configure an OLE DB source component.
IDTSComponentMetaData90 source = dataFlowTask.ComponentMetaDataCollection.New();
source.ComponentClassID = "DTSAdapter.OLEDBSource.1";
// Create the design-time instance of the source.
CManagedComponentWrapper srcDesignTime = source.Instantiate();
// The ProvideComponentProperties method creates a default output.
srcDesignTime.ProvideComponentProperties();
source.Name = "Employee Dim from OLTP";
// Assign the connection manager.
source.RuntimeConnectionCollection[0].ConnectionManagerID = SconMgr.ID;
source.RuntimeConnectionCollection[0].ConnectionManager = DtsConvert.ToConnectionManager90(SconMgr);
// Set the custom properties of the source.
srcDesignTime.SetComponentProperty("AccessMode", 0); // Mode 0 : OpenRowset / Table - View
srcDesignTime.SetComponentProperty("OpenRowset", "[dbo].[Employee_Dim]");
// Connect to the data source, and then update the metadata for the source.
srcDesignTime.AcquireConnections(null);
srcDesignTime.ReinitializeMetaData();
srcDesignTime.ReleaseConnections();
// Add an Row Count to the data flow.
IDTSComponentMetaData90 RowCountComponent = dataFlowTask.ComponentMetaDataCollection.New();
RowCountComponent.Name = "Row Count";
RowCountComponent.ComponentClassID = "DTSTransform.RowCount.1"; //Public Token is: "{DE50D3C7-41AF-4804-9247-CF1DEB147971}";
CManagedComponentWrapper rowCountDesignTime = RowCountComponent.Instantiate();
rowCountDesignTime.ProvideComponentProperties();
rowCountDesignTime.SetComponentProperty("VariableName", "RowCountVar");
rowCountDesignTime.AcquireConnections(null);
rowCountDesignTime.ReinitializeMetaData();
rowCountDesignTime.ReleaseConnections();
// Create the path from source to Row Count Transformation.
IDTSPath90 pathSource_RowCount = dataFlowTask.PathCollection.New();
pathSource_RowCount.AttachPathAndPropagateNotifications(source.OutputCollection[0], RowCountComponent.InputCollection[0]);
// Create and configure an OLE DB destination component.
IDTSComponentMetaData90 destination = dataFlowTask.ComponentMetaDataCollection.New();
destination.ComponentClassID = "DTSAdapter.OLEDBDestination.1";
// Create the design-time instance of the destination.
CManagedComponentWrapper destDesignTime = destination.Instantiate();
// The ProvideComponentProperties method creates a default input.
destDesignTime.ProvideComponentProperties();
destination.Name = "Employee Dim from OLAP";
// Assign the connection manager.
destination.RuntimeConnectionCollection[0].ConnectionManagerID = DconMgr.ID;
destination.RuntimeConnectionCollection[0].ConnectionManager = DtsConvert.ToConnectionManager90(DconMgr);
// Set the custom properties.
destDesignTime.SetComponentProperty("AccessMode", 3); // Mode 3 : OpenRowset Using FastLoad / Table - View fast load
destDesignTime.SetComponentProperty("OpenRowset", "[EmployeeDim]");
destDesignTime.AcquireConnections(null);
destDesignTime.ReinitializeMetaData();
destDesignTime.ReleaseConnections();
// Create the path from Rowcount to destination.
IDTSPath90 pathRowCount_Dest = dataFlowTask.PathCollection.New();
pathRowCount_Dest.AttachPathAndPropagateNotifications(RowCountComponent.OutputCollection[0], destination.InputCollection[0]);
// Get the destination's default input and virtual input.
IDTSInput90 input = destination.InputCollection[0];
IDTSVirtualInput90 vInput = input.GetVirtualInput();
// Iterate through the virtual input column collection.
foreach (IDTSVirtualInputColumn90 vColumn in vInput.VirtualInputColumnCollection)
{
// Call the SetUsageType method of the destination
// to add each available virtual input column as an input column.
destDesignTime.SetUsageType(
input.ID, vInput, vColumn.LineageID, DTSUsageType.UT_READONLY);
}
//map external metadata to the inputcolumn
int index = 0;
foreach (IDTSInputColumn90 inputColumn in input.InputColumnCollection)
{
IDTSExternalMetadataColumn90 exMetaColumn = (IDTSExternalMetadataColumn90)input.ExternalMetadataColumnCollection[index++];
inputColumn.ExternalMetadataColumnID = exMetaColumn.ID;
}
//-----------Data Flow Inner component ends-------------------
// Precedence constraints from "Package Start Indicator" to "Load Employee Dim"
PrecedenceConstraint PackageStartIndicatorTask_dataFlowTaskHost = package.PrecedenceConstraints.Add(package.Executables[@"Package Start Indicator"], package.Executables[@"Load Employee Dim"]);
PackageStartIndicatorTask_dataFlowTaskHost.Name = @"Package Start Indicator" + @"Load Employee Dim";
//To add Package End Indicator (Execute Sql task )
TaskHost PackageEndIndicatorTask = (TaskHost)package.Executables.Add("Microsoft.SqlServer.Dts.Tasks.ExecuteSQLTask.ExecuteSQLTask, Microsoft.SqlServer.SQLTask, Version=9.0.242.0, Culture=neutral, PublicKeyToken=89845dcd8080cc91");
PackageEndIndicatorTask.Name = @"Package End Indicator";
PackageEndIndicatorTask.Description = @"Execute SQL Task Description";
XmlDocument PackageEndIndicatorTaskdoc = new XmlDocument();
PackageEndIndicatorTaskdoc.LoadXml(@"");
((IDTSComponentPersist)PackageEndIndicatorTask.InnerObject).LoadFromXML(PackageEndIndicatorTaskdoc.DocumentElement, null);
// Precedence constraints from "Load Employee Dim" to "Package End Indicator"
PrecedenceConstraint dataFlowTaskHost_PackageEndIndicatorTask = package.PrecedenceConstraints.Add(package.Executables[@"Load Employee Dim"], package.Executables[@"Package End Indicator"]);
dataFlowTaskHost_PackageEndIndicatorTask.Name = @"Load Employee Dim" + @"Package End Indicator";
//Saving the package
Microsoft.SqlServer.Dts.Runtime.Application app = new Microsoft.SqlServer.Dts.Runtime.Application();
app.SaveToXml(@"I:Site workSamplePackageSamplePackageSamplePackage.dtsx", package, null);
}
}
}
VB .Net Expand
Imports System
Imports System
Imports System.Collections.Generic
Imports System.Text
Imports Microsoft.SqlServer.Dts.Runtime
Imports Microsoft.SqlServer.Dts.Pipeline.Wrapper
Imports System.Data.SqlClient
Imports System.Xml
'The below code will create a SSIS package for SQL Server 2005
Module Module1
Sub Main(ByVal args As String())
'To Create a package named [Sample Package]
Dim package As New Package()
package.Name = "Sample Package"
package.PackageType = DTSPackageType.DTSDesigner90
package.VersionBuild = 1
'To add Connection Manager to the package
'For source database (OLTP)
Dim OLTP As ConnectionManager = package.Connections.Add("OLEDB")
OLTP.ConnectionString = "Data Source=.;Initial Catalog=OLTP;Provider=SQLNCLI.1;Integrated Security=SSPI;Auto Translate=False;"
OLTP.Name = "LocalHost.OLTP"
'For destination database (OLAP)
Dim OLAP As ConnectionManager = package.Connections.Add("OLEDB")
OLAP.ConnectionString = "Data Source=.;Initial Catalog=OLAP;Provider=SQLNCLI.1;Integrated Security=SSPI;Auto Translate=False;"
OLAP.Name = "LocalHost.OLAP"
'To add package variables
Dim TransactionID As Variable = package.Variables.Add("TransactionID", False, "User", 0)
TransactionID.Name = "TransactionID"
Dim RowCountVar As Variable = package.Variables.Add("RowCountVar", False, "User", 0)
RowCountVar.Name = "RowCountVar"
'To add Package Start Indicator (Execute Sql task )
Dim PackageStartIndicatorTask As TaskHost = DirectCast(package.Executables.Add("Microsoft.SqlServer.Dts.Tasks.ExecuteSQLTask.ExecuteSQLTask, Microsoft.SqlServer.SQLTask, Version=9.0.242.0, Culture=neutral, PublicKeyToken=89845dcd8080cc91"), TaskHost)
PackageStartIndicatorTask.Name = "Package Start Indicator"
PackageStartIndicatorTask.Description = "Execute SQL Task Description"
Dim PackageStartIndicatorTaskdoc As New XmlDocument()
PackageStartIndicatorTaskdoc.LoadXml("")
DirectCast(PackageStartIndicatorTask.InnerObject, IDTSComponentPersist).LoadFromXML(PackageStartIndicatorTaskdoc.DocumentElement, Nothing)
'To add Load Employee Dim to the package [Data Flow Task]
Dim dataFlowTaskHost As TaskHost = DirectCast(package.Executables.Add("DTS.Pipeline.1"), TaskHost)
dataFlowTaskHost.Name = "Load Employee Dim"
dataFlowTaskHost.FailPackageOnFailure = True
dataFlowTaskHost.FailParentOnFailure = True
dataFlowTaskHost.DelayValidation = False
dataFlowTaskHost.Description = "Data Flow Task"
'-----------Data Flow Inner component starts----------------
Dim dataFlowTask As MainPipe = TryCast(dataFlowTaskHost.InnerObject, MainPipe)
' Source OLE DB connection manager to the package.
Dim SconMgr As ConnectionManager = package.Connections("LocalHost.OLTP")
' Destination OLE DB connection manager to the package.
Dim DconMgr As ConnectionManager = package.Connections("LocalHost.OLAP")
' Create and configure an OLE DB source component.
Dim source As IDTSComponentMetaData90 = dataFlowTask.ComponentMetaDataCollection.[New]()
source.ComponentClassID = "DTSAdapter.OLEDBSource.1"
' Create the design-time instance of the source.
Dim srcDesignTime As CManagedComponentWrapper = source.Instantiate()
' The ProvideComponentProperties method creates a default output.
srcDesignTime.ProvideComponentProperties()
source.Name = "Employee Dim from OLTP"
' Assign the connection manager.
source.RuntimeConnectionCollection(0).ConnectionManagerID = SconMgr.ID
source.RuntimeConnectionCollection(0).ConnectionManager = DtsConvert.ToConnectionManager90(SconMgr)
' Set the custom properties of the source.
srcDesignTime.SetComponentProperty("AccessMode", 0)
' Mode 0 : OpenRowset / Table - View
srcDesignTime.SetComponentProperty("OpenRowset", "[dbo].[Employee_Dim]")
' Connect to the data source, and then update the metadata for the source.
srcDesignTime.AcquireConnections(Nothing)
srcDesignTime.ReinitializeMetaData()
srcDesignTime.ReleaseConnections()
' Add an Row Count to the data flow.
Dim RowCountComponent As IDTSComponentMetaData90 = dataFlowTask.ComponentMetaDataCollection.[New]()
RowCountComponent.Name = "Row Count"
RowCountComponent.ComponentClassID = "DTSTransform.RowCount.1"
'Public Token is: "{DE50D3C7-41AF-4804-9247-CF1DEB147971}";
Dim rowCountDesignTime As CManagedComponentWrapper = RowCountComponent.Instantiate()
rowCountDesignTime.ProvideComponentProperties()
rowCountDesignTime.SetComponentProperty("VariableName", "RowCountVar")
rowCountDesignTime.AcquireConnections(Nothing)
rowCountDesignTime.ReinitializeMetaData()
rowCountDesignTime.ReleaseConnections()
' Create the path from source to Row Count Transformation.
Dim pathSource_RowCount As IDTSPath90 = dataFlowTask.PathCollection.[New]()
pathSource_RowCount.AttachPathAndPropagateNotifications(source.OutputCollection(0), RowCountComponent.InputCollection(0))
' Create and configure an OLE DB destination component.
Dim destination As IDTSComponentMetaData90 = dataFlowTask.ComponentMetaDataCollection.[New]()
destination.ComponentClassID = "DTSAdapter.OLEDBDestination.1"
' Create the design-time instance of the destination.
Dim destDesignTime As CManagedComponentWrapper = destination.Instantiate()
' The ProvideComponentProperties method creates a default input.
destDesignTime.ProvideComponentProperties()
destination.Name = "Employee Dim from OLAP"
' Assign the connection manager.
destination.RuntimeConnectionCollection(0).ConnectionManagerID = DconMgr.ID
destination.RuntimeConnectionCollection(0).ConnectionManager = DtsConvert.ToConnectionManager90(DconMgr)
' Set the custom properties.
destDesignTime.SetComponentProperty("AccessMode", 3)
' Mode 3 : OpenRowset Using FastLoad / Table - View fast load
destDesignTime.SetComponentProperty("OpenRowset", "[EmployeeDim]")
destDesignTime.AcquireConnections(Nothing)
destDesignTime.ReinitializeMetaData()
destDesignTime.ReleaseConnections()
' Create the path from Rowcount to destination.
Dim pathRowCount_Dest As IDTSPath90 = dataFlowTask.PathCollection.[New]()
pathRowCount_Dest.AttachPathAndPropagateNotifications(RowCountComponent.OutputCollection(0), destination.InputCollection(0))
' Get the destination's default input and virtual input.
Dim input As IDTSInput90 = destination.InputCollection(0)
Dim vInput As IDTSVirtualInput90 = input.GetVirtualInput()
' Iterate through the virtual input column collection.
For Each vColumn As IDTSVirtualInputColumn90 In vInput.VirtualInputColumnCollection
' Call the SetUsageType method of the destination
' to add each available virtual input column as an input column.
destDesignTime.SetUsageType(input.ID, vInput, vColumn.LineageID, DTSUsageType.UT_READONLY)
Next
'map external metadata to the inputcolumn
Dim index As Integer = 0
For Each inputColumn As IDTSInputColumn90 In input.InputColumnCollection
Dim exMetaColumn As IDTSExternalMetadataColumn90 = DirectCast(input.ExternalMetadataColumnCollection(index), IDTSExternalMetadataColumn90)
index = index + 1
inputColumn.ExternalMetadataColumnID = exMetaColumn.ID
Next
'-----------Data Flow Inner component ends-------------------
' Precedence constraints from "Package Start Indicator" to "Load Employee Dim"
Dim PackageStartIndicatorTask_dataFlowTaskHost As PrecedenceConstraint = package.PrecedenceConstraints.Add(package.Executables("Package Start Indicator"), package.Executables("Load Employee Dim"))
PackageStartIndicatorTask_dataFlowTaskHost.Name = "Package Start Indicator" & "Load Employee Dim"
'To add Package End Indicator (Execute Sql task )
Dim PackageEndIndicatorTask As TaskHost = DirectCast(package.Executables.Add("Microsoft.SqlServer.Dts.Tasks.ExecuteSQLTask.ExecuteSQLTask, Microsoft.SqlServer.SQLTask, Version=9.0.242.0, Culture=neutral, PublicKeyToken=89845dcd8080cc91"), TaskHost)
PackageEndIndicatorTask.Name = "Package End Indicator"
PackageEndIndicatorTask.Description = "Execute SQL Task Description"
Dim PackageEndIndicatorTaskdoc As New XmlDocument()
PackageEndIndicatorTaskdoc.LoadXml("")
DirectCast(PackageEndIndicatorTask.InnerObject, IDTSComponentPersist).LoadFromXML(PackageEndIndicatorTaskdoc.DocumentElement, Nothing)
' Precedence constraints from "Load Employee Dim" to "Package End Indicator"
Dim dataFlowTaskHost_PackageEndIndicatorTask As PrecedenceConstraint = package.PrecedenceConstraints.Add(package.Executables("Load Employee Dim"), package.Executables("Package End Indicator"))
dataFlowTaskHost_PackageEndIndicatorTask.Name = "Load Employee Dim" & "Package End Indicator"
'Saving the package
Dim app As New Microsoft.SqlServer.Dts.Runtime.Application()
app.SaveToXml("I:Site workSamplePackageSamplePackageSamplePackage.dtsx", package, Nothing)
End Sub
End Module
Programming Integration Service in SQL Server 2008
C# .Net Expand
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.SqlServer.Dts.Runtime;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using System.Data.SqlClient;
using System.Xml;
namespace SamplePackageCSharp2008
{
class Program
{
static void Main(string[] args)
{
//To Create a package named [Sample Package]
Package package = new Package();
package.Name = "Sample Package";
package.PackageType = DTSPackageType.DTSDesigner100;
package.VersionBuild = 1;
//To add Connection Manager to the package
//For source database (OLTP)
ConnectionManager OLTP = package.Connections.Add("OLEDB");
OLTP.ConnectionString = "Data Source=.;Initial Catalog=OLTP;Provider=SQLNCLI10;Integrated Security=SSPI;Auto Translate=False;";
OLTP.Name = "LocalHost.OLTP";
//For destination database (OLAP)
ConnectionManager OLAP = package.Connections.Add("OLEDB");
OLAP.ConnectionString = "Data Source=.;Initial Catalog=OLAP;Provider=SQLNCLI10;Integrated Security=SSPI;Auto Translate=False;";
OLAP.Name = "LocalHost.OLAP";
//To add package variables
Variable TransactionID = package.Variables.Add("TransactionID", false, "User", 0);
TransactionID.Name = @"TransactionID";
Variable RowCountVar = package.Variables.Add("RowCountVar", false, "User", 0);
RowCountVar.Name = @"RowCountVar";
//To add Package Start Indicator (Execute Sql task )
TaskHost PackageStartIndicatorTask = (TaskHost)package.Executables.Add("Microsoft.SqlServer.Dts.Tasks.ExecuteSQLTask.ExecuteSQLTask, Microsoft.SqlServer.SQLTask, Version=10.0.0.0, Culture=neutral, PublicKeyToken=89845dcd8080cc91");
PackageStartIndicatorTask.Name = @"Package Start Indicator";
PackageStartIndicatorTask.Description = @"Execute SQL Task Description";
XmlDocument PackageStartIndicatorTaskdoc = new XmlDocument();
PackageStartIndicatorTaskdoc.LoadXml(@"");
((IDTSComponentPersist)PackageStartIndicatorTask.InnerObject).LoadFromXML(PackageStartIndicatorTaskdoc.DocumentElement, null);
//To add Load Employee Dim to the package [Data Flow Task]
TaskHost dataFlowTaskHost = (TaskHost)package.Executables.Add("SSIS.Pipeline.2");
dataFlowTaskHost.Name = @"Load Employee Dim";
dataFlowTaskHost.FailPackageOnFailure = true;
dataFlowTaskHost.FailParentOnFailure = true;
dataFlowTaskHost.DelayValidation = false;
dataFlowTaskHost.Description = @"Data Flow Task";
//-----------Data Flow Inner component starts----------------
MainPipe dataFlowTask = dataFlowTaskHost.InnerObject as MainPipe;
// Source OLE DB connection manager to the package.
ConnectionManager SconMgr = package.Connections["LocalHost.OLTP"];
// Destination OLE DB connection manager to the package.
ConnectionManager DconMgr = package.Connections["LocalHost.OLAP"];
// Create and configure an OLE DB source component.
IDTSComponentMetaData100 source = dataFlowTask.ComponentMetaDataCollection.New();
source.ComponentClassID = "DTSAdapter.OLEDBSource.2";
// Create the design-time instance of the source.
CManagedComponentWrapper srcDesignTime = source.Instantiate();
// The ProvideComponentProperties method creates a default output.
srcDesignTime.ProvideComponentProperties();
source.Name = "Employee Dim from OLTP";
// Assign the connection manager.
source.RuntimeConnectionCollection[0].ConnectionManagerID = SconMgr.ID;
source.RuntimeConnectionCollection[0].ConnectionManager = DtsConvert.GetExtendedInterface(SconMgr);
// Set the custom properties of the source.
srcDesignTime.SetComponentProperty("AccessMode", 0); // Mode 0 : OpenRowset / Table - View
srcDesignTime.SetComponentProperty("OpenRowset", "[dbo].[Employee_Dim]");
// Connect to the data source, and then update the metadata for the source.
srcDesignTime.AcquireConnections(null);
srcDesignTime.ReinitializeMetaData();
srcDesignTime.ReleaseConnections();
// Add an Row Count to the data flow.
IDTSComponentMetaData100 RowCountComponent = dataFlowTask.ComponentMetaDataCollection.New();
RowCountComponent.Name = "Row Count";
RowCountComponent.ComponentClassID = "DTSTransform.RowCount.2";
CManagedComponentWrapper rowCountDesignTime = RowCountComponent.Instantiate();
rowCountDesignTime.ProvideComponentProperties();
rowCountDesignTime.SetComponentProperty("VariableName", "RowCountVar");
rowCountDesignTime.AcquireConnections(null);
rowCountDesignTime.ReinitializeMetaData();
rowCountDesignTime.ReleaseConnections();
// Create the path from source to Row Count Transformation.
IDTSPath100 pathSource_RowCount = dataFlowTask.PathCollection.New();
pathSource_RowCount.AttachPathAndPropagateNotifications(source.OutputCollection[0], RowCountComponent.InputCollection[0]);
// Create and configure an OLE DB destination component.
IDTSComponentMetaData100 destination = dataFlowTask.ComponentMetaDataCollection.New();
destination.ComponentClassID = "DTSAdapter.OLEDBDestination.2";
// Create the design-time instance of the destination.
CManagedComponentWrapper destDesignTime = destination.Instantiate();
// The ProvideComponentProperties method creates a default input.
destDesignTime.ProvideComponentProperties();
destination.Name = "Employee Dim from OLAP";
// Assign the connection manager.
destination.RuntimeConnectionCollection[0].ConnectionManagerID = DconMgr.ID;
destination.RuntimeConnectionCollection[0].ConnectionManager = DtsConvert.GetExtendedInterface(DconMgr);
// Set the custom properties.
destDesignTime.SetComponentProperty("AccessMode", 3); // Mode 3 : OpenRowset Using FastLoad / Table - View fast load
destDesignTime.SetComponentProperty("OpenRowset", "[EmployeeDim]");
destDesignTime.AcquireConnections(null);
destDesignTime.ReinitializeMetaData();
destDesignTime.ReleaseConnections();
// Create the path from Rowcount to destination.
IDTSPath100 pathRowCount_Dest = dataFlowTask.PathCollection.New();
pathRowCount_Dest.AttachPathAndPropagateNotifications(RowCountComponent.OutputCollection[0], destination.InputCollection[0]);
// Get the destination's default input and virtual input.
IDTSInput100 input = destination.InputCollection[0];
IDTSVirtualInput100 vInput = input.GetVirtualInput();
// Iterate through the virtual input column collection.
foreach (IDTSVirtualInputColumn100 vColumn in vInput.VirtualInputColumnCollection)
{
// Call the SetUsageType method of the destination
// to add each available virtual input column as an input column.
destDesignTime.SetUsageType(
input.ID, vInput, vColumn.LineageID, DTSUsageType.UT_READONLY);
}
//map external metadata to the inputcolumn
int index = 0;
foreach (IDTSInputColumn100 inputColumn in input.InputColumnCollection)
{
IDTSExternalMetadataColumn100 exMetaColumn = (IDTSExternalMetadataColumn100)input.ExternalMetadataColumnCollection[index++];
inputColumn.ExternalMetadataColumnID = exMetaColumn.ID;
}
//-----------Data Flow Inner component ends-------------------
// Precedence constraints from "Package Start Indicator" to "Load Employee Dim"
PrecedenceConstraint PackageStartIndicatorTask_dataFlowTaskHost = package.PrecedenceConstraints.Add(package.Executables[@"Package Start Indicator"], package.Executables[@"Load Employee Dim"]);
PackageStartIndicatorTask_dataFlowTaskHost.Name = @"Package Start Indicator" + @"Load Employee Dim";
//To add Package End Indicator (Execute Sql task )
TaskHost PackageEndIndicatorTask = (TaskHost)package.Executables.Add("Microsoft.SqlServer.Dts.Tasks.ExecuteSQLTask.ExecuteSQLTask, Microsoft.SqlServer.SQLTask, Version=10.0.0.0, Culture=neutral, PublicKeyToken=89845dcd8080cc91");
PackageEndIndicatorTask.Name = @"Package End Indicator";
PackageEndIndicatorTask.Description = @"Execute SQL Task Description";
XmlDocument PackageEndIndicatorTaskdoc = new XmlDocument();
PackageEndIndicatorTaskdoc.LoadXml(@"");
((IDTSComponentPersist)PackageEndIndicatorTask.InnerObject).LoadFromXML(PackageEndIndicatorTaskdoc.DocumentElement, null);
// Precedence constraints from "Load Employee Dim" to "Package End Indicator"
PrecedenceConstraint dataFlowTaskHost_PackageEndIndicatorTask = package.PrecedenceConstraints.Add(package.Executables[@"Load Employee Dim"], package.Executables[@"Package End Indicator"]);
dataFlowTaskHost_PackageEndIndicatorTask.Name = @"Load Employee Dim" + @"Package End Indicator";
//Saving the package
Microsoft.SqlServer.Dts.Runtime.Application app = new Microsoft.SqlServer.Dts.Runtime.Application();
app.SaveToXml(@"I:Site workSamplePackage 2008SamplePackageSamplePackage.dtsx", package, null);
}
}
}
VB .Net Expand
Imports System
Imports System.Collections.Generic
Imports System.Linq
Imports System.Text
Imports Microsoft.SqlServer.Dts.Runtime
Imports Microsoft.SqlServer.Dts.Pipeline.Wrapper
Imports System.Data.SqlClient
Imports System.Xml
'The below code will create a SSIS package for SQL Server 2008
Module Module1
Sub Main()
'To Create a package named [Sample Package]
Dim package As New Package()
package.Name = "Sample Package"
package.PackageType = DTSPackageType.DTSDesigner100
package.VersionBuild = 1
'To add Connection Manager to the package
'For source database (OLTP)
Dim OLTP As ConnectionManager = package.Connections.Add("OLEDB")
OLTP.ConnectionString = "Data Source=.;Initial Catalog=OLTP;Provider=SQLNCLI10;Integrated Security=SSPI;Auto Translate=False;"
OLTP.Name = "LocalHost.OLTP"
'For destination database (OLAP)
Dim OLAP As ConnectionManager = package.Connections.Add("OLEDB")
OLAP.ConnectionString = "Data Source=.;Initial Catalog=OLAP;Provider=SQLNCLI10;Integrated Security=SSPI;Auto Translate=False;"
OLAP.Name = "LocalHost.OLAP"
'To add package variables
Dim TransactionID As Variable = package.Variables.Add("TransactionID", False, "User", 0)
TransactionID.Name = "TransactionID"
Dim RowCountVar As Variable = package.Variables.Add("RowCountVar", False, "User", 0)
RowCountVar.Name = "RowCountVar"
'To add Package Start Indicator (Execute Sql task )
Dim PackageStartIndicatorTask As TaskHost = DirectCast(package.Executables.Add("Microsoft.SqlServer.Dts.Tasks.ExecuteSQLTask.ExecuteSQLTask, Microsoft.SqlServer.SQLTask, Version=10.0.0.0, Culture=neutral, PublicKeyToken=89845dcd8080cc91"), TaskHost)
PackageStartIndicatorTask.Name = "Package Start Indicator"
PackageStartIndicatorTask.Description = "Execute SQL Task Description"
Dim PackageStartIndicatorTaskdoc As New XmlDocument()
PackageStartIndicatorTaskdoc.LoadXml("")
DirectCast(PackageStartIndicatorTask.InnerObject, IDTSComponentPersist).LoadFromXML(PackageStartIndicatorTaskdoc.DocumentElement, Nothing)
'To add Load Employee Dim to the package [Data Flow Task]
Dim dataFlowTaskHost As TaskHost = DirectCast(package.Executables.Add("SSIS.Pipeline.2"), TaskHost)
dataFlowTaskHost.Name = "Load Employee Dim"
dataFlowTaskHost.FailPackageOnFailure = True
dataFlowTaskHost.FailParentOnFailure = True
dataFlowTaskHost.DelayValidation = False
dataFlowTaskHost.Description = "Data Flow Task"
'-----------Data Flow Inner component starts----------------
Dim dataFlowTask As MainPipe = TryCast(dataFlowTaskHost.InnerObject, MainPipe)
' Source OLE DB connection manager to the package.
Dim SconMgr As ConnectionManager = package.Connections("LocalHost.OLTP")
' Destination OLE DB connection manager to the package.
Dim DconMgr As ConnectionManager = package.Connections("LocalHost.OLAP")
' Create and configure an OLE DB source component.
Dim source As IDTSComponentMetaData100 = dataFlowTask.ComponentMetaDataCollection.[New]()
source.ComponentClassID = "DTSAdapter.OLEDBSource.2"
' Create the design-time instance of the source.
Dim srcDesignTime As CManagedComponentWrapper = source.Instantiate()
' The ProvideComponentProperties method creates a default output.
srcDesignTime.ProvideComponentProperties()
source.Name = "Employee Dim from OLTP"
' Assign the connection manager.
source.RuntimeConnectionCollection(0).ConnectionManagerID = SconMgr.ID
source.RuntimeConnectionCollection(0).ConnectionManager = DtsConvert.GetExtendedInterface(SconMgr)
' Set the custom properties of the source.
srcDesignTime.SetComponentProperty("AccessMode", 0)
' Mode 0 : OpenRowset / Table - View
srcDesignTime.SetComponentProperty("OpenRowset", "[dbo].[Employee_Dim]")
' Connect to the data source, and then update the metadata for the source.
srcDesignTime.AcquireConnections(Nothing)
srcDesignTime.ReinitializeMetaData()
srcDesignTime.ReleaseConnections()
' Add an Row Count to the data flow.
Dim RowCountComponent As IDTSComponentMetaData100 = dataFlowTask.ComponentMetaDataCollection.[New]()
RowCountComponent.Name = "Row Count"
RowCountComponent.ComponentClassID = "DTSTransform.RowCount.2"
Dim rowCountDesignTime As CManagedComponentWrapper = RowCountComponent.Instantiate()
rowCountDesignTime.ProvideComponentProperties()
rowCountDesignTime.SetComponentProperty("VariableName", "RowCountVar")
rowCountDesignTime.AcquireConnections(Nothing)
rowCountDesignTime.ReinitializeMetaData()
rowCountDesignTime.ReleaseConnections()
' Create the path from source to Row Count Transformation.
Dim pathSource_RowCount As IDTSPath100 = dataFlowTask.PathCollection.[New]()
pathSource_RowCount.AttachPathAndPropagateNotifications(source.OutputCollection(0), RowCountComponent.InputCollection(0))
' Create and configure an OLE DB destination component.
Dim destination As IDTSComponentMetaData100 = dataFlowTask.ComponentMetaDataCollection.[New]()
destination.ComponentClassID = "DTSAdapter.OLEDBDestination.2"
' Create the design-time instance of the destination.
Dim destDesignTime As CManagedComponentWrapper = destination.Instantiate()
' The ProvideComponentProperties method creates a default input.
destDesignTime.ProvideComponentProperties()
destination.Name = "Employee Dim from OLAP"
' Assign the connection manager.
destination.RuntimeConnectionCollection(0).ConnectionManagerID = DconMgr.ID
destination.RuntimeConnectionCollection(0).ConnectionManager = DtsConvert.GetExtendedInterface(DconMgr)
' Set the custom properties.
destDesignTime.SetComponentProperty("AccessMode", 3)
' Mode 3 : OpenRowset Using FastLoad / Table - View fast load
destDesignTime.SetComponentProperty("OpenRowset", "[EmployeeDim]")
destDesignTime.AcquireConnections(Nothing)
destDesignTime.ReinitializeMetaData()
destDesignTime.ReleaseConnections()
' Create the path from Rowcount to destination.
Dim pathRowCount_Dest As IDTSPath100 = dataFlowTask.PathCollection.[New]()
pathRowCount_Dest.AttachPathAndPropagateNotifications(RowCountComponent.OutputCollection(0), destination.InputCollection(0))
' Get the destination's default input and virtual input.
Dim input As IDTSInput100 = destination.InputCollection(0)
Dim vInput As IDTSVirtualInput100 = input.GetVirtualInput()
' Iterate through the virtual input column collection.
For Each vColumn As IDTSVirtualInputColumn100 In vInput.VirtualInputColumnCollection
' Call the SetUsageType method of the destination
' to add each available virtual input column as an input column.
destDesignTime.SetUsageType(input.ID, vInput, vColumn.LineageID, DTSUsageType.UT_READONLY)
Next
'map external metadata to the inputcolumn
Dim index As Integer = 0
For Each inputColumn As IDTSInputColumn100 In input.InputColumnCollection
Dim exMetaColumn As IDTSExternalMetadataColumn100 = DirectCast(input.ExternalMetadataColumnCollection(index), IDTSExternalMetadataColumn100)
index = index + 1
inputColumn.ExternalMetadataColumnID = exMetaColumn.ID
Next
'-----------Data Flow Inner component ends-------------------
' Precedence constraints from "Package Start Indicator" to "Load Employee Dim"
Dim PackageStartIndicatorTask_dataFlowTaskHost As PrecedenceConstraint = package.PrecedenceConstraints.Add(package.Executables("Package Start Indicator"), package.Executables("Load Employee Dim"))
PackageStartIndicatorTask_dataFlowTaskHost.Name = "Package Start Indicator" & "Load Employee Dim"
'To add Package End Indicator (Execute Sql task )
Dim PackageEndIndicatorTask As TaskHost = DirectCast(package.Executables.Add("Microsoft.SqlServer.Dts.Tasks.ExecuteSQLTask.ExecuteSQLTask, Microsoft.SqlServer.SQLTask, Version=10.0.0.0, Culture=neutral, PublicKeyToken=89845dcd8080cc91"), TaskHost)
PackageEndIndicatorTask.Name = "Package End Indicator"
PackageEndIndicatorTask.Description = "Execute SQL Task Description"
Dim PackageEndIndicatorTaskdoc As New XmlDocument()
PackageEndIndicatorTaskdoc.LoadXml("")
DirectCast(PackageEndIndicatorTask.InnerObject, IDTSComponentPersist).LoadFromXML(PackageEndIndicatorTaskdoc.DocumentElement, Nothing)
' Precedence constraints from "Load Employee Dim" to "Package End Indicator"
Dim dataFlowTaskHost_PackageEndIndicatorTask As PrecedenceConstraint = package.PrecedenceConstraints.Add(package.Executables("Load Employee Dim"), package.Executables("Package End Indicator"))
dataFlowTaskHost_PackageEndIndicatorTask.Name = "Load Employee Dim" & "Package End Indicator"
'Saving the package
Dim app As New Microsoft.SqlServer.Dts.Runtime.Application()
app.SaveToXml("I:Site workSamplePackage 2008SamplePackageSamplePackage.dtsx", package, Nothing)
End Sub
End Module
Like the above we can use any transformation and task in building a package programmatically.
To make your work more easy and effortless,
SQL Lion team comes up with a unique tool named “
SSIS Component Explorer” that will help you in retrieving the
Creation Name, Component Class ID, etc about any component like Control Flow Tasks, Data Flow Transformations, Connections Managers, Log Providers, etc in SSIS package. It also provides code snippets for each component.
Download the free version from
here.
Hi. I’m trying to test your code, but the debugger raise an exception on line
PackageStartIndicatorTaskdoc.LoadXml(@”");
The error is:
System.Xml.XmlException was unhandled
Root element is missing.
Bye