Creating a Meta Data Driven SSIS Solution with Biml – 3. Creating a Table Package

In this third blogpost on creating a meta data driven ssis solution I’ll create a biml document with which you can create a package that will import the contents of a query in the source database to a table in the destination database. In blogpost 4 I will enhance this document to include biml script that loops through the entries in the meta data table SsisPackages and create a package for every entry  in this table.

The design of the TablePackage is loosely based on the design used by Joy Mundy and Warren Thornthwaite in their book:
The Microsoft Data warehouse Toolkit With SQL Server 2008 R2. So it will include a system for logging results, adding lineage to the imported rows in the data warehouse and creates error files if appropriate.

The Control Flow

The control flow consist of:

  1. An Execute SQL Task, retrieving the number of rows in the destination table at the start of the package.
  2. (Optional, depends on a setting in metadata table) An Execute SQL Task, Truncating the destination table.
  3. An Execute SQL statement to the meta database, creating a new row in the SsisAudit table and returning the AuditID of the created row. Which is used for Lineage in the Data Flow.
  4. The Data Flow, see next paragraph
  5. An Execute SQL Task, retrieving the number of rows in the destination table at the end of the package.
  6. An Execute SQL Task to update the audit row with the end time and several row counters.
The Data Flow

The data flow will first try to import the data with a bulk-load setting (step 4). It will try to load any error rows using a row by row setting. (step5) Any remaining errors rows will be written to a raw text file.(step 7) In the data flow we will also count the extracted rows(step 2) and any error rows (step 6). The package will also add the meta data (AuditID and time stamps) to the data flow (step 3)

The Biml code

<Biml xmlns="">
        <OleDbConnection Name="metaDB"
                         ConnectionString="Data Source=.;Initial Catalog=MyDwh_meta;Provider=SQLNCLI10.1;Integrated Security=SSPI;"
        <OleDbConnection Name="Northwind"
                         ConnectionString="Data Source=.;Initial Catalog=Northwind;Provider=SQLNCLI10.1;Integrated Security=SSPI;"
        <OleDbConnection Name="stagingDB"
                         ConnectionString="Data Source=.;Initial Catalog=MyDwh_st;Provider=SQLNCLI10.1;Integrated Security=SSPI;"
        <OleDbConnection Name="DWH"
                         ConnectionString="Data Source=.;Initial Catalog=MyDwh;Provider=SQLNCLI10.1;Integrated Security=SSPI;"
        <Package Name="Import_Northwind_Categories" ConstraintMode="Linear" AutoCreateConfigurationsType="None">
                <Variable Name="AuditID" DataType="Int32" Namespace="User" >0</Variable>
                <Variable Name="PackageID" DataType="Int32" Namespace="User" >99</Variable>
                <Variable Name="RcStart" DataType="Int32" Namespace="User" >0</Variable>
                <Variable Name="RcExtract" DataType="Int32" Namespace="User" >0</Variable>
                <Variable Name="RcInsert" DataType="Int32" Namespace="User" >0</Variable>
                <Variable Name="RcUpdate" DataType="Int32" Namespace="User" >0</Variable>
                <Variable Name="RcError" DataType="Int32" Namespace="User">0</Variable>
                <Variable Name="RcEnd" DataType="Int32" Namespace="User">0</Variable>
                <Variable Name="ErrorFile" DataType="String" Namespace="User">c:\myErrorFile.raw</Variable>
            <!--Package Configurations-->
                <PackageConfiguration Name="MetaDB" >
                    <EnvironmentVariableInput EnvironmentVariable="Northwind_Config"></EnvironmentVariableInput>
                        <ConfigurationValue DataType="String"
                                            Value="Data Source=.;Initial Catalog=ROCWB_MIS_config;Provider=SQLNCLI10.1;Integrated Security=SSPI;">
                <PackageConfiguration Name="Northwind" ConnectionName="metaDB">
                    <ExternalTableInput Table="[dbo].[SsisConfiguration]" />
                <PackageConfiguration Name="stagingDB" ConnectionName="metaDB">
                    <ExternalTableInput Table="[dbo].[SsisConfiguration]" />
                <PackageConfiguration Name="DWH" ConnectionName="metaDB">
                    <ExternalTableInput Table="[dbo].[SsisConfiguration]" />
                <!--Get Inital RowCount-->
                <ExecuteSQL Name="SQL - Get Initial RowCount" ConnectionName="stagingDB" ResultSet="SingleRow">
                    <DirectInput>SELECT COUNT(*) AS RcStart FROM [dbo].[Northwind_Categories]</DirectInput>
                        <Result Name="RcStart" VariableName="User.RcStart" />
                <!--Truncate Table-->
                <ExecuteSQL Name="SQL - Truncate Destination Table" ConnectionName="stagingDB" ResultSet="None">
                    <DirectInput>TRUNCATE TABLE [dbo].[Northwind_Categories]</DirectInput>
                <!--Get auditID-->
                <ExecuteSQL Name="SQL - Create AuditRow and Get AuditID" ConnectionName="metaDB" ResultSet="SingleRow">
                    <DirectInput>EXEC [dbo].[uspNewAuditRow] ?,?,?,?,?,?</DirectInput>
                        <Parameter Name="0" Direction="Input" DataType="Int32" VariableName="User.PackageID" />
                        <Parameter Name="1" Direction="Input" DataType="Guid" VariableName="System.PackageID" />
                        <Parameter Name="2" Direction="Input" DataType="Guid" VariableName="System.VersionGUID" />
                        <Parameter Name="3" Direction="Input" DataType="Guid" VariableName="System.ExecutionInstanceGUID" />
                        <Parameter Name="4" Direction="Input" DataType="DateTime" VariableName="System.StartTime" />
                        <Parameter Name="5" Direction="Input" DataType="Int32" VariableName="User.RcStart" />
                        <Result Name="0" VariableName="User.AuditID" />
                <!--Import DIM_KOSTENPLAATS-->
                <Dataflow Name="DFT - Import_Northwind_Categories">
                        <OleDbSource Name="ODS - Northwind" ConnectionName="Northwind">
                            <DirectInput>SELECT CategoryID, CategoryName FROM [dbo].[Categories]</DirectInput>
                        <RowCount Name="RC - Extracted Rows"  VariableName="User.RcExtract" />
                        <DerivedColumns Name="DC - Add Metadata">
                                <Column Name="InsertAuditID" DataType="Int32" ReplaceExisting="false">@AuditID</Column>
                                <Column Name="UpdateAuditID" DataType="Int32" ReplaceExisting="false">@AuditID</Column>
                                <Column Name="RowStartDate" DataType="DateTime" ReplaceExisting="false">@StartTime</Column>
                                <Column Name="RowEndDate" DataType="DateTime" ReplaceExisting="false">(DT_dbDATE)"12/31/9999"</Column>
                        <OleDbDestination Name="ODD-BulkLoad stagingDB" ConnectionName="stagingDB">
                            <ExternalTableOutput Table="[dbo].[Northwind_Categories]"/>
                            <ErrorHandling ErrorRowDisposition="RedirectRow" TruncationRowDisposition="RedirectRow" />
                        <OleDbDestination Name="ODD-RowbyRow stagingDB" ConnectionName="stagingDB" UseFastLoadIfAvailable="false" >
                            <InputPath OutputPathName="ODD-BulkLoad stagingDB.Error" />
                            <ExternalTableOutput Table="[dbo].[Northwind_Categories]"/>
                            <ErrorHandling ErrorRowDisposition="RedirectRow" TruncationRowDisposition="RedirectRow" />
                        <RowCount Name="RC - Error Rows" VariableName="User.RcError">
                            <InputPath OutputPathName="ODD-RowbyRow stagingDB.Error" />
                        <RawFileDestination Name="RTF - Error File">
                            <FileFromVariableOutput VariableName="User.ErrorFile"/>
                <!--Get Final RowCount-->
                <ExecuteSQL Name="SQL - Get Final RowCount" ConnectionName="stagingDB" ResultSet="SingleRow">
                    <DirectInput>SELECT COUNT(*) AS RcEnd FROM [dbo].[Northwind_Categories]</DirectInput>
                        <Result Name="RcEnd" VariableName="User.RcEnd" />
                <!--Update Audit Row-->
                <ExecuteSQL Name="SQL - Update Audit Row" ConnectionName="metaDB" ResultSet="None">
                    <DirectInput>EXEC [dbo].[uspUpdateAuditRow] ?,?,?,?,?,?</DirectInput>
                        <Parameter Name="0" Direction="Input" DataType="Int32" VariableName="User.AuditID" />
                        <Parameter Name="1" Direction="Input" DataType="Int32" VariableName="User.RcExtract" />
                        <Parameter Name="2" Direction="Input" DataType="Int32" VariableName="User.RcInsert" />
                        <Parameter Name="3" Direction="Input" DataType="Int32" VariableName="User.RcUpdate" />
                        <Parameter Name="4" Direction="Input" DataType="Int32" VariableName="User.RcError" />
                        <Parameter Name="5" Direction="Input" DataType="Int32" VariableName="User.RcEnd" />

7. Create a package – and test it.
  • Open Business Intelligence Development Studio and create a new Integration Services Project.
  • Right click  the project in solution explorer and select Add New Biml File.
  • Rename the file to: Create a package.biml
  • Copy the Biml code on this page to the newly created biml.file.
  • Right click the Create a package.biml in Solution Explorer and select Generate SSIS Packages.
  • Open the newly created package Import_Northwind_Categories.dtsx and examine the contents.
  • Test the package by Executing the package