Search results matching tags 'Design Pattern' and 'ETL Instrumentation'

SSIS Design Patterns, the Book

andyleonard — Mon, 06 Aug 2012 16:37:43 GMT

For the past two years, I have had the honor and privilege or authoring SSIS Design Patterns alongside Jessica Moss, Michelle Ufford, Tim Mitchell, and Matt Masson. Publication of the book – like many projects of this scope – has been delayed. The current publication date is 27 Aug 2012 and I have high confidence in this date.

I take responsibility for publication delays and apologize to those who pre-ordered the book. The reasons for the delays are not important. I have built a career as a software developer and architect based on the following maxim:

Deliver quality late, no one remembers.
Deliver junk on time, no one forgets.

The shared goal of everyone working on this project has been to deliver quality. Proofing the manuscripts, I believe we have achieved that goal.

:{>

SSIS Design Pattern - ETL Instrumentation, Part 4

andyleonard — Wed, 26 Dec 2007 07:50:00 GMT

Introduction

This post is part of a series of posts on ETL Instrumentation.

In Part 1 we built a database to hold collected SSIS run time metrics and an SSIS package to deomnstrate how and why we would load metrics into the database.

In Part 2 we expanded on our database and the SSIS package to annotate version metadata, manage error metrics capture, and task status reporting.

In Part 3, we started using the ETL Instrumentation infrastructure we have built to measure some actual ETL. We started by counting rows.

In Part 4, we continue instrumenting by adding yet another ETL process and again scaling our measurement capabilities.

A Brief History Of Our ETL Instrumentation Project

To review, our metrics database is named SSISRunTimeMetrics. It contains a schema named ssis. In this schema are eleven objects:
- a table named ssis.RunTimeMetrics.
- a table named ssis.RunTimeErrors.
- a table named ssis.TaskMetrics.
- a table named ssis.RowCounts.
- a table named ssis.RowCountTypes.
- a stored procedure named ssis.usp_RecordPackageStart.
- a stored procedure named ssis.usp_RecordPackageEnd.
- a stored procedure named ssis.usp_RecordPackageError.
- a stored procedure named ssis.usp_RecordTaskStart.
- a stored procedure named ssis.usp_RecordTaskEnd.
- a stored procedure named ssis.usp_RecordRowCounts.

Our source database is AdventureWorks and our destination database is SSISRunTimeMetrics_Target. SSISRunTimeMetrics_Target contains one object:
- a table named dbo.Contact.

We expanded the types of run-time data we are collecting. Part 1 introduced Status collection, in Part 2 we added Exception collection. We also introduced scope into both types of collection, recording Exception information on error and finalizing Status (reporting that an error occurred).

At the beginning of SSIS package execution, we call ssis.usp_RecordPackageStart from an Execute SQL Task. We pass the package start date and time, the package name, and the package version. We also pass in a status of "Started". From this stored procedure we get the ID of the newly created row, which we then push into a Package Load ID variable (iPackageLoadID).

At the beginning of a task or collection of tasks that define a process, we call ssis.usp_RecordTaskStart from an Execute SQL Task. We pass the task or process start date and time, the task (source) name, iPackageLoadID, and a status of "Started". From this stored procedure we get the ID of the newly created row, which we then push into a Task Load ID variable (iTaskLoadID).

We have a Data Flow Task to move rows from the AdventureWorks.Person.Contact table to a target database and table we created: SSISRunTimeMetrics_Target.dbo.Contact. We optimized the package for set-based updates and collect row count metrics which are inserted into SSISRunTimeMetrics.ssis.usp_RecordRowCounts.

When this task completes, we call ssis.usp_RecordTaskEnd from an Execute SQL Task. We pass in the Task Load ID from the iTaskLoadID variable, the current date and time, and the status "Succeeded".

On error, we capture Exception data and record an Error Status - both are crucial to knowing what happens when an exception is thrown.

When the package completes execution, we call ssis.usp_RecordPackageEnd from an Execute SQL Task. We pass in the Package Load ID from the variable, the current date and time, and the status "Succeeded".

Let's get started on the next step!

Version Control

First, update version information:

Remember to update Version properties:

Now we are ready to start developing.

Sell, Sell, Sell

Let's extract and load some Sales data.

Open the SSISRunTimeMetrics package you built previously. Delete the Success Precedence Constraint between the "Step 1 - Load Contact" Sequence Container and the "Log End of Package Execution" Execute SQL Task.

Drag a Sequence Container onto the Control Flow canvas. Move the "Log End of Package Execution" Execute SQL Task down some and position the new Sequence Container between the "Step 1 - Load Contact" Sequence Container and the "Log End of Package Execution" Execute SQL Task.

Connect the "Step 1 - Load Contact" Sequence Container to the new Sequence Container with a Success Precedence Constraint, and the new Sequence Container to the "Log End of Package Execution" Execute SQL Task with a Success Precedence Constraint.

Rename the new Sequence Container "Step 2 - Load Sales".

Good design is reuseable. Maybe not 100%, but most good designs are at least partially reuseable. Such is the case here - we have a good design in "Step 1 - Load Contact" - we will reuse lots of it in "Step 2 - Load Sales". Let's frame-out the flow, then fill in the details.

Drag two Execute SQL Tasks and a Data Flow Task into "Step 2 - Load Sales".

Name the first Execute SQL Task "Load Sales" and double-click it to open the editor. Set the ResultSet property to "Single row" and the Connection property to "(local).SSISRunTimeMetrics". Enter the following in the SQLStatement property:

declare @Now datetime
set @Now = GetDate()
exec ssis.usp_RecordTaskStart ?,NULL,@Now,?,'Started'

On the Parameter Mappings page, add two input parameters. Set Parameter 0 to Long data type and supply the User::iPackageLoadID variable. Set Parameter 1 to VarChar data type and supply the System::TaskName variable:

On the Result Set page, add one Result named 0 aimed at the User::iTaskLoadID variable:

Click the OK button to close the editor.

Before proceeding, note that this Execute SQL Task is also the product of good design. In fact, the only difference task and it's counterpart in "Step 1 - Load Contact" is the name of the task itself. Everything else is identical.

So why not copy and paste the task? Good question - we certainly could have! And we will copy and paste other tasks starting now.

Connect a Success Precedence Constraint from the "Load Sales" Execute SQL Task to the Data Flow Task. We need to do some cleanup here before proceeding. In the "Step 1 - Load Contact" Sequence Container there's a Data Flow Task named "Data Flow Task". We have one of those in our "Step 2 - Load Sales" Sequence Container as well. This is permissible because the objects are in different containers and have different scope.

It robs us of an important navigation facility - one we will likely need: the ability to use the Data Flow Task tab's dropdown box. Have a look:

To remedy this, let's rename the Data Flow Task in "Step 1 - Load Contact" "Load Contact Data". Similary, let's rename the "Step 2 - Load Sales" Data Flow Task "Load Sales Data".

There. Much better.

Connect the "Load Sales Data" Data Flow Task to the second Execute SQL Task with a Success Precedence Constraint and rename it (the second Execute SQL Task) "Apply Staged Updates". Double-click it to open the editor and set the Connection property to "(local).SSISRunTimeMetrics_Target". We will return to this task later - click the OK button to close the editor.

Copy the "Log Successful End of Task" Execute SQL Task from the "Step 1 - Load Contact" Sequence Container and paste it onto the Data Flow canvas. Then drag it into the the "Step 2 - Load Sales" Sequence Container.

Note: You can paste it directly into the "Step 2 - Load Sales" Sequence Container if you want to, but I recommend you not do this in SSIS 2005. The Sequence Container will expand to accomodate anything inside it, and the paste functionality in SSIS 2005 completely ignores the mouse pointer position (and eveything else, so far as I can tell) when you paste from the clipboard. Combined, these behaviors cause sequence containers to grow unpredictably large when you paste directly into them.

No modifications are required for the "Log Successful End of Task" Execute SQL Task to function as desired in the new sequence container - how cool is that?

Copy the "Log Failed End Of Task" Execute SQL Task and paste it onto the Control Flow canvas. The new task will show up named "Log Failed End Of Task 1". Again, a naming convention conflict. To resolve it, rename the original "Log Failed End Of Task" Execute SQL Task - connected to the "Step 1 - Load Contact" Sequence Container via a Failure Precedence Constraint - to "Log Failed End of Load Contact Task".

Rename the newly pasted "Log Failed End Of Task 1" Execute SQL Task to "Log Failed End Of Load Sales Task" and connect the "Step 2 - Load Sales" Sequence Container to "Log Failed End Of Load Sales Task" via a Failure Precedence Constraint.

Copy the "Record Row Count" Execute SQL Task from the "Step 1 - Load Contact" Sequence Container. Again, paste it onto the Control Flow canvas and then drag it into the "Step 2 - Load Sales" Sequence Container. Connect the "Log Successful End Of Task" Execute SQL Task to the "Record Row Count" Execute SQL Task with a Success Precedence Constraint and double-click the task to open the editor.

All is well with the General page, but the Parameter Mapping page reveals some poor variable-naming choices in the last exercise. We can fix this in the variable dropdown. Click the dropdown that currently contains the User::iContactCount variable and select <New variable...>:

When the Add Variable dialog displays, click the Container dropdown and select the package ("SSISRunTimeMetrics"). This determines the scope of the variable and we want a package-scoped variable.

Click the OK button to select the package scope. Set the Name of the variable to iSalesInputCount, the Value Type (data type) to Int32, and the Value to 0:

Click the OK button to close the Add Variable dialog.

Repeat the procedure above for the "Counts" variables. Name the remaining three Counts variables iSalesNewRowsCount, iSalesChangedRowsCount, and iSalesUnchangedRowsCount; respectively. When complete, the Parameter Mapping page should appear as shown:

Click the OK button to close the Execute SQL Task editor.

The flow is now framed-out. We are ready to begin our Sales-specific coding.

Building The LZ (Landing Zone)

We need a place for our Sales data to land in SSISRunTimeMetrics_Target database.

In this section I am going to walk through the first phase of the process of converting a well-designed OLTP schema into a denormalized schema.

We'll start with the AdventureWorks Sales schema. First, let's list all the tables in the Sales schema using the following query:

use AdventureWorks;
go

select s.name + '.' + t.name
from sys.tables t
inner join sys.schemas s on s.schema_id = t.schema_id
where s.name = 'Sales'

This gives us a list of tables in the Sales schema:

Sales.StoreContact
Sales.ContactCreditCard
Sales.CountryRegionCurrency
Sales.CreditCard
Sales.Currency
Sales.SalesOrderDetail
Sales.CurrencyRate
Sales.Customer
Sales.SalesOrderHeader
Sales.CustomerAddress
Sales.SalesOrderHeaderSalesReason
Sales.SalesPerson
Sales.SalesPersonQuotaHistory
Sales.SalesReason
Sales.Individual
Sales.SalesTaxRate
Sales.SalesTerritory
Sales.SalesTerritoryHistory
Sales.ShoppingCartItem
Sales.SpecialOffer
Sales.SpecialOfferProduct
Sales.Store

Let's select Sales.SalesOrderDetail as our base table... we have to start somewhere. Open SQL Server Management Studio and connect the Object Browser to your local (or development) instance of SQL Server 2005. Expand Databases, then AdventureWorks, then Tables:

Scroll down to Sales.SalesOrderDetail. Right-click the table object in Object Browser. Hover over "Script Table as", then "CREATE To", and click "New Query Editor Window":

This creates a nice CREATE script (and more) for the Sales.SalesOrderDetail table. I only need the CREATE TABLE portion so I remove the rest. I modify the script further - making the table part of the dbo schema. I discard the constraints, NOT NULLs, brackets, and extended properties and I'm left with:

CREATE TABLE dbo.SalesOrderDetail(
SalesOrderID int NULL,
SalesOrderDetailID int NULL,
CarrierTrackingNumber nvarchar(25) NULL,
OrderQty smallint NULL,
ProductID int NULL,
SpecialOfferID int NULL,
UnitPrice money NULL,
UnitPriceDiscount money NULL,
LineTotal money,
rowguid uniqueidentifier,
ModifiedDate datetime NULL)

Repeating the process for the Sale.SalesOrderHeader table yields:

CREATE TABLE dbo.SalesOrderHeader(
SalesOrderID int NULL,
RevisionNumber tinyint NULL,
OrderDate datetime NULL,
DueDate datetime NULL,
ShipDate datetime NULL,
Status tinyint NULL,
OnlineOrderFlag bit NULL,
SalesOrderNumber nvarchar(25) NULL,
PurchaseOrderNumber nvarchar(25) NULL,
AccountNumber nvarchar(15) NULL,
CustomerID int NULL,
ContactID int NULL,
SalesPersonID int NULL,
TerritoryID int NULL
BillToAddressID int NULL,
ShipToAddressID int NULL,
ShipMethodID int NULL,
CreditCardID int NULL,
CreditCardApprovalCode varchar(15) NULL,
CurrencyRateID int NULL,
SubTotal money NULL,
TaxAmt money NULL,
Freight money NULL,
TotalDue money NULL,
Comment nvarchar(128) NULL,
rowguid uniqueidentifier NULL,
ModifiedDate datetime NULL)

I can now combine these statements, removing the duplication, to create a destination table statement:

CREATE TABLE dbo.SalesOrderHeaderDetail(
SalesOrderID int NULL,
SalesOrderDetailID int NULL,
CarrierTrackingNumber nvarchar(25) NULL,
OrderQty smallint NULL,
ProductID int NULL,
SpecialOfferID int NULL,
UnitPrice money NULL,
UnitPriceDiscount money NULL,
LineTotal money,
SalesOrderDtatilrowguid uniqueidentifier,
SalesOrderDetailModifiedDate datetime NULL,
RevisionNumber tinyint NULL,
OrderDate datetime NULL,
DueDate datetime NULL,
ShipDate datetime NULL,
Status tinyint NULL,
OnlineOrderFlag bit NULL,
SalesOrderNumber nvarchar(25) NULL,
PurchaseOrderNumber nvarchar(25) NULL,
AccountNumber nvarchar(15) NULL,
CustomerID int NULL,
ContactID int NULL,
SalesPersonID int NULL,
TerritoryID int NULL,
BillToAddressID int NULL,
ShipToAddressID int NULL,
ShipMethodID int NULL,
CreditCardID int NULL,
CreditCardApprovalCode varchar(15) NULL,
CurrencyRateID int NULL,
SubTotal money NULL,
TaxAmt money NULL,
Freight money NULL,
TotalDue money NULL,
Comment nvarchar(128) NULL,
SalesOrderHeaderrowguid uniqueidentifier NULL,
SalesOrderHeaderModifiedDate datetime NULL)

Execute this statement against the SSISRunTimeMetrics_Target database to create our destination table.

Filling In The Blanks

Double-click the "Load Sales Data" Data Flow Task to switch to the Data Flow tab for editing. Drag an OLE DB Source Adapter onto the canvas and double-click it to open the editor. Select the (local).AdventureWorks connection manager. Change the Data access mode to Sql Command and enter the following SQL statement into the SQL Command Text textbox:

SELECT SalesOrderID
,SalesOrderDetailID
,CarrierTrackingNumber
,OrderQty
,ProductID
,SpecialOfferID
,UnitPrice
,UnitPriceDiscount
,LineTotal
,rowguid
,ModifiedDate
FROM Sales.SalesOrderDetail

Click the OK button to close the editor. Right-click the Source Adapter and rename it "Sales Detail Source":

Drag a second OLE DB Source Adapter onto the Data Flow canvas and double-click it to open the editor. Select "(local).AdventureWorks" as the connection manager and SQL Command as the Data Access Mode. Enter the following statement into the SQL Command Text textbox:

SELECT SalesOrderID
,RevisionNumber
,OrderDate
,DueDate
,ShipDate
,Status
,OnlineOrderFlag
,SalesOrderNumber
,PurchaseOrderNumber
,AccountNumber
,CustomerID
,ContactID
,SalesPersonID
,TerritoryID
,BillToAddressID
,ShipToAddressID
,ShipMethodID
,CreditCardID
,CreditCardApprovalCode
,CurrencyRateID
,SubTotal
,TaxAmt
,Freight
,TotalDue
,Comment
,rowguid
,ModifiedDate
FROM Sales.SalesOrderHeader

Click the OK button to close the editor and rename the Source Adapter "Sales Header Source".

Drag a Merge Join onto the Data Flow canvas and connect a Data Flow Path (green arrow) from each Source Adapter to the Merge Join:

Note the Merge Join has an error - the Left Input is not sorted. (Neither is the Right Input, but validation fails on, and reports, the first error). To address this condition, right-click each Source Adapter and select Show Advanced Editor:

Click on the Input and Output Properties tab and expand the OLE DB Source Output object, then expand the Output Columns logical folder. Click on the OLE DB Source Output object (which represents the Output buffer) and change the IsSorted property to True:

In the Output Columns list, click on the SalesOrderID column and change the SortKeyPosition property to 1:

Click the OK button to close the Advanced Editor. Double-click the source adapter to open the editor and append an Order By clause to the SQL Command: "ORDER BY SalesOrderID". Close the editor and repeat this process for the other Source Adapter.

But wait - we still have an error:

Double-click the Merge Join to open the editor. Click every column from the Left input and every column except SalesOrderID from the Right Input. Two columns are named the same in both tables - rowguid and ModifiedDate. To differentiate, prepend each column's Output Alias with the table name:

Click the OK button to close the editor. The error clears.

We did all that to set this up.

Count 'Em Up

As with the Contacts data, we will count the rows entering the process. Depending on the nature of the process, source data, and desired measurement(s); you may choose to measure immediately after the Source Adapters or after the Merge Join - or both. We'll start with the same kind of counts measurements we built in the Contacts data flow.

Drag a Row Count transformation onto the Data Flow canvas and connect the output of the Merge Join to its input. Double-click the Row Count to open the editor and assign the User::iSalesInputCount to the VariableName property:

As with the Contacts data, our next steps are to correlate and filter the data, so drag a Lookup and Conditional Split transformation onto the data flow canvas and connect them (in respective order) to the Row Count transformation:

Double-click the Lookup to open the editor and assign the following properties:
OLE DB Connection Manager: (local).SSISRunTimeMetrics_Target
Table or View: dbo.SalesOrdeHeaderDetail

Click the Columns tab, right-click in the white-space, and click Select All Mappings. Right-click again and select Delete Selected Mappings. Connect the SalesOrderID and SalesOrderDetailID columns. Select every column in the Available Lookup Columns list by checking each checkbox, then prepend each Output Alias with "Dest_":

Click the Configure Error Output button and change the Lookup Error from "Fail Component" to "Ignore Failure". Remember, this converts the default INNER JOIN functionality of the Lookup transformation into a LEFT OUTER JOIN. Click the OK button to close the Error Output Configuration, then click the OK button again to close the Lookup editor.

We are loading the pipeline with lookup data in our data flow that matches data - by SalesOrderID and SalesOrderDetailID - in the destination.

Connect the output data flow path of the Lookup of the Lookup transformation to the Conditional Split transformation and double-click the Conditional Split transformation to open the editor. Create a new output named "New Sales" with the Condition: "IsNull(Dest_SalesOrderDetailID)". If the LEFT OUTER JOIN functionality of the Lookup returns a NULL Dest_SalesOrderDetailID - and really every destination column will be NULL if there's no matching destination row, we could use any of them - then this is a new Sales data row.

Add a second condition named "Changed Sales" with the following condition expression:

(ISNULL(Dest_RevisionNumber) ? -1 : Dest_RevisionNumber) != (ISNULL(RevisionNumber) ? -1 : RevisionNumber) || (ISNULL(Dest_OrderDate) ? (DT_DBDate)0 : Dest_OrderDate) != (ISNULL(OrderDate) ? (DT_DBDate)0 : OrderDate) || (ISNULL(Dest_DueDate) ? (DT_DBDate)0 : Dest_DueDate) != (ISNULL(DueDate) ? (DT_DBDate)0 : DueDate) || (ISNULL(Dest_ShipDate) ? (DT_DBDate)0 : Dest_ShipDate) != (ISNULL(ShipDate) ? (DT_DBDate)0 : ShipDate) || (ISNULL(Dest_Status) ? 0 : Dest_Status) != (ISNULL(Status) ? 0 : Status) || (ISNULL(Dest_OnlineOrderFlag) ? TRUE : Dest_OnlineOrderFlag) != (ISNULL(OnlineOrderFlag) ? TRUE : OnlineOrderFlag) || (ISNULL(Dest_SalesOrderNumber) ? "NULL" : Dest_SalesOrderNumber) != (ISNULL(SalesOrderNumber) ? "NULL" : SalesOrderNumber) || (ISNULL(Dest_PurchaseOrderNumber) ? "NULL" : Dest_PurchaseOrderNumber) != (ISNULL(PurchaseOrderNumber) ? "NULL" : PurchaseOrderNumber) || (ISNULL(Dest_AccountNumber) ? "NULL" : Dest_AccountNumber) != (ISNULL(AccountNumber) ? "NULL" : AccountNumber) || (ISNULL(Dest_CustomerID) ? -1 : Dest_CustomerID) != (ISNULL(CustomerID) ? -1 : CustomerID) || (ISNULL(Dest_ContactID) ? -1 : Dest_ContactID) != (ISNULL(ContactID) ? -1 : ContactID) || (ISNULL(Dest_SalesPersonID) ? -1 : Dest_SalesPersonID) != (ISNULL(SalesPersonID) ? -1 : SalesPersonID) || (ISNULL(Dest_TerritoryID) ? -1 : Dest_TerritoryID) != (ISNULL(TerritoryID) ? -1 : TerritoryID) || (ISNULL(Dest_BillToAddressID) ? -1 : Dest_BillToAddressID) != (ISNULL(BillToAddressID) ? -1 : BillToAddressID) || (ISNULL(Dest_ShipToAddressID) ? -1 : Dest_ShipToAddressID) != (ISNULL(ShipToAddressID) ? -1 : ShipToAddressID) || (ISNULL(Dest_ShipMethodID) ? -1 : Dest_ShipMethodID) != (ISNULL(ShipMethodID) ? -1 : ShipMethodID) || (ISNULL(Dest_CreditCardID) ? -1 : Dest_CreditCardID) != (ISNULL(CreditCardID) ? -1 : CreditCardID) || (ISNULL(Dest_CreditCardApprovalCode) ? "NULL" : Dest_CreditCardApprovalCode) != (ISNULL(CreditCardApprovalCode) ? "NULL" : CreditCardApprovalCode) || (ISNULL(Dest_CurrencyRateID) ? -1 : Dest_CurrencyRateID) != (ISNULL(CurrencyRateID) ? -1 : CurrencyRateID) || (ISNULL(Dest_SubTotal) ? 0 : Dest_SubTotal) != (ISNULL(SubTotal) ? 0 : SubTotal) || (ISNULL(Dest_TaxAmt) ? 0 : Dest_TaxAmt) != (ISNULL(TaxAmt) ? 0 : TaxAmt) || (ISNULL(Dest_Freight) ? 0 : Dest_Freight) != (ISNULL(Freight) ? 0 : Freight) || (ISNULL(Dest_TotalDue) ? 0 : Dest_TotalDue) != (ISNULL(TotalDue) ? 0 : TotalDue) || (ISNULL(Dest_Comment) ? "NULL" : Dest_Comment) != (ISNULL(Comment) ? "NULL" : Comment) || (ISNULL(Dest_SalesOrderHeaderModifiedDate) ? (DT_DBDATE)0 : Dest_SalesOrderHeaderModifiedDate) != (ISNULL(SalesOrderHeaderModifiedDate) ? (DT_DBDATE)0 : SalesOrderHeaderModifiedDate) || (ISNULL(Dest_CarrierTrackingNumber) ? "NULL" : Dest_CarrierTrackingNumber) != (ISNULL(CarrierTrackingNumber) ? "NULL" : CarrierTrackingNumber) || (ISNULL(Dest_OrderQty) ? 0 : Dest_OrderQty) != (ISNULL(OrderQty) ? 0 : OrderQty) || (ISNULL(Dest_ProductID) ? -1 : Dest_ProductID) != (ISNULL(ProductID) ? -1 : ProductID) || (ISNULL(Dest_SpecialOfferID) ? -1 : Dest_SpecialOfferID) != (ISNULL(SpecialOfferID) ? -1 : SpecialOfferID)

Rename the default output "Unchanged Rows". Click the OK button to close the editor.

Drag an OLE DB Destination Adapter onto the Data Flow canvas and rename it "New Sales Destination". Connect an output of the Conditional Split to the new Destination Adapter. When prompted, select the "New Sales" output of the Conditional Split:

Double-click the Destination Adapter ot open the editor. Set the Connection Manager property to (local).SSISRunTimeMetrics_Target. Set the Data Access Mode property to "Table or View" and select the dbo.SalesOrderHeaderDetail table. Click on the Mappings page to automap the pipeline fields to the table columns:

Click the OK button to close the editor.

Drag another OLE DB Destination Adapter onto the Data Flow canvas and rename it "stgSalesChangedRows". Connect an output from the Conditional Split to the new Destination Adapter and select the "Changed Sales" output when prompted. Double-click the Destination Adapter to open the editor. Set the Connection Manager property to (local).SSISRunTimeMetrics_Target and set the Data Access Mode property to "Table or View". Click the New button next to the "Name of the Table or View" dropdown:

Click the OK button to create the stgSalesChangedRows table. Click the Mappings page to automap the columns, then click the OK button to close the editor.

We now have Sales ETL.

To complete our counts logic, add three Row Count transformations to the Data Flow canvas. Name them "New Rows Count", "Changed Rows Count", and "Unchanged Rows Count". Position "New Rows Count" between the Conditional Split and the "New Sales Destination" Adapter. Double-click to open the editor and set the VariableName property to "User::iSalesNewRowsCount". Click the OK button to close the editor.

Position the "Changed Rows Count" between the Conditional Split and the stgSalesChangedRows Destination Adapter. Open its editor and set the VariableName property to "User::iSalesChangedRowsCount".

Open the editor for the "Unchanged Rows Count" transformation and set the VariableName property to "User::iSalesUnchangedRowsCount".

Before we leave this section, let's complete the staged updates by returning to the Control Flow and updating the SQLStatement property of the "Apply Staged Updates" Execute SQL Task inside the "Step 2 - Load Sales" Sequence Container with the following statement:

UPDATE dest
SET dest.SalesOrderID = stage.SalesOrderID
,dest.CarrierTrackingNumber = stage.CarrierTrackingNumber
,dest.OrderQty = stage.OrderQty
,dest.ProductID = stage.ProductID
,dest.SpecialOfferID = stage.SpecialOfferID
,dest.UnitPrice = stage.UnitPrice
,dest.UnitPriceDiscount = stage.UnitPriceDiscount
,dest.LineTotal = stage.LineTotal
,dest.SalesOrderDtatilrowguid = stage.SalesOrderDtatilrowguid
,dest.SalesOrderDetailModifiedDate = stage.SalesOrderDetailModifiedDate
,dest.RevisionNumber = stage.RevisionNumber
,dest.OrderDate = stage.OrderDate
,dest.DueDate = stage.DueDate
,dest.ShipDate = stage.ShipDate
,dest.Status = stage.Status
,dest.OnlineOrderFlag = stage.OnlineOrderFlag
,dest.SalesOrderNumber = stage.SalesOrderNumber
,dest.PurchaseOrderNumber = stage.PurchaseOrderNumber
,dest.AccountNumber = stage.AccountNumber
,dest.CustomerID = stage.CustomerID
,dest.ContactID = stage.ContactID
,dest.SalesPersonID = stage.SalesPersonID
,dest.TerritoryID = stage.TerritoryID
,dest.BillToAddressID = stage.BillToAddressID
,dest.ShipToAddressID = stage.ShipToAddressID
,dest.ShipMethodID = stage.ShipMethodID
,dest.CreditCardID = stage.CreditCardID
,dest.CreditCardApprovalCode = stage.CreditCardApprovalCode
,dest.CurrencyRateID = stage.CurrencyRateID
,dest.SubTotal = stage.SubTotal
,dest.TaxAmt = stage.TaxAmt
,dest.Freight = stage.Freight
,dest.TotalDue = stage.TotalDue
,dest.Comment = stage.Comment
,dest.SalesOrderHeaderrowguid = stage.SalesOrderHeaderrowguid
,dest.SalesOrderHeaderModifiedDate = stage.SalesOrderHeaderModifiedDate
FROM dbo.SalesOrderHeaderDetail dest
INNER JOIN dbo.stgSalesChangedRows stage ON stage.SalesOrderDetailID = dest.SalesOrderDetailID

There. Done and done.

If You Build It...

Up until now, we've basically followed the same template used for Contacts to construct the Sales ETL. We added some complexity (by design) to grow our understanding of ETL along with our knowledge of SSIS.

Our ETL measurement is record counts and record counts only. Let's expand on that some by also capturing a monetary sum. This will add even more confidence in our ETL, once we validate (Validation is Part 5).

Let's begin by creating a new destination table to hold our sums: SSISRunTimeMetrics.ssis.RowSums. Use the following script to create the table:

This table is remarkably similar to the ssis.RowCounts table we created to hold Row Count data - and for good reason, the functions of these two tables are remarkably similar. As with the Row Counts data, we need to add a stored procedure to insert Sums data, and another table to hold Inserted Types... or do we? Instead of re-creating the functionality contained in the ssis.RowCountTypes table, let's rename - and expand the purpose of - the table.

Executing the following script accomplishes this nicely:

use SSISRunTimeMetrics
go

-- vars...
declare @sql varchar(255)

-- create ssis schema...
if not exists(select name
              from sys.schemas
              where name = 'ssis')
begin
set @sql = 'Create Schema ssis'
exec(@sql)
end

-- delete RowCountTypes table, if exists...
if exists(select s.name + '.' + t.name
          from sys.tables t
          inner join sys.schemas s on s.schema_id = t.schema_id
          where t.name = 'RowCountTypes'
           and s.name = 'ssis')
drop table ssis.RowCountTypes
go

-- delete RowTypes table, if exists...
if exists(select s.name + '.' + t.name
          from sys.tables t
          inner join sys.schemas s on s.schema_id = t.schema_id
          where t.name = 'RowTypes'
           and s.name = 'ssis')
drop table ssis.RowTypes
go

Create Table ssis.RowTypes
(RowTypeID char(1) not null
,RowTypeName varchar(25) null
,RowTypeDescription varchar(255) null)
go

if not exists(select RowTypeID
              from ssis.RowTypes
              where RowTypeID = 'I')
insert into ssis.RowTypes
(RowTypeID
,RowTypeName
,RowTypeDescription)
values
('I'
,'Selected Input Rows'
,'Input rows selected from a source')

if not exists(select RowTypeID
              from ssis.RowTypes
              where RowTypeID = 'N')
insert into ssis.RowTypes
(RowTypeID
,RowTypeName
,RowTypeDescription)
values
('N'
,'New Rows'
,'New rows')

if not exists(select RowTypeID
              from ssis.RowTypes
              where RowTypeID = 'C')
insert into ssis.RowTypes
(RowTypeID
,RowTypeName
,RowTypeDescription)
values
('C'
,'Changed Rows'
,'Changed rows')

if not exists(select RowTypeID
              from ssis.RowTypes
              where RowTypeID = 'U')
insert into ssis.RowTypes
(RowTypeID
,RowTypeName
,RowTypeDescription)
values
('U'
,'Unchanged Rows'
,'No changes detected in rows')
go

Our stored procedure to accomplish inserts:

use SSISRunTimeMetrics
go
if exists(select s.name + '.' + p.name
from sys.procedures p
inner join sys.schemas s on s.schema_id = p.schema_id
where p.name = 'usp_RecordRowSum'
and s.name = 'ssis')
begin
Drop Procedure ssis.usp_RecordRowSum
end
go
Create Procedure ssis.usp_RecordRowSum
@RunTimeMetricsID int
,@TaskMetricsID int
,@RowSum decimal(38,2)
,@RowSumTypeID char(1)
,@RowSumColumnName varchar(255) = null
,@ParentTaskMetricsID int = null
As
begin
-- insert the run time errors data...
insert into ssis.RowSums
(TaskMetricsID
,RunTimeMetricsId
,ParentTaskMetricsID
,RowSum
,RowSumColumnName
,RowSumTypeID)
values
(@TaskMetricsID
,@RunTimeMetricsID
,@ParentTaskMetricsID
,@RowSum
,@RowSumColumnName
,@RowSumTypeID)
end

go

Now that our infrastructure is built we can start using it to load SSIS run time metrics.

Add 'Em Up

We need variables to hold the sums we intend to collect. Right-click the Control Flow and click Variables. Click the New Variable button and add a package-scoped variable, data type Double, named iSalesInputAmount. Repeat the process for three other Double variables named iSalesNewAmount, iSalesChangedAmount, and iSalesUnchangedAmount.

There are other ways to load this type of data. The way I choose to demonstrate here is not the cleanest but it clearly exercises the principles of ETL Instrumentation.

Return to the "Load Sales Data" Data Flow Task and add one each Multicast and Aggregate transformations. Position the Mulitcast between Merge Join and Input Row Count transformations and connect them through it. Rename the Aggregate transformation "Input Line Total" and connect another output of the Multicast to it:

Double-click the Aggregate transformation to open the editor and check the LineTotal input column. Select Sum from the Operation column - this will add the total of all the LineTotal columns that pass between the Merge Join and Input Row Count transformations:

Add a Script Component to the Data Flow. When prompted for Script Component Type, select Destination:

Rename the Script Component "Push InputLineTotal into Input Variable", connect the "Input Line Total" Aggregate transformation to it, and double-click the Script Component to open the editor.

On the Input Columns page, check the Line Total input. On the Script page, enter iSalesInputAmount in the ReadWriteVariables property and click the Design Script button. In the script editor, enter the following code:

Dim iAmount As Double

Public Overrides Sub Input0_ProcessInputRow(ByVal Row As Input0Buffer)
'
' Add your code here
'

iAmount = Row.LineTotal

End Sub

Public Overrides Sub PostExecute()

MyBase.PostExecute()
Me.Variables.iSalesInputAmount = iAmount
End Sub

Close the Script Editor and click the OK button to clase the Script Component editor. This should load the aggregated value into the iSalesInputAmount variable.

Drag two each Multicast and three each Aggregate, and Script Component transformations onto the Data Flow canvas. Repeat the procedure outlined above for the New, Changed, and Unchanged Conditional Split outputs - for the iSalesNewAmount, iSalesChangedAmount, and iSalesUnchangedAmount variable values (respectively). Note you do not need a Multicase transformation for the Unchanged output. Sum the LineTotal fields for each output.

This is a lot of work and there is lots of room for error. Take your time. Double-check your work. Don't take shortcuts. When complete, the New section will look something like this:

Load 'Em Up

Return to the Control Flow - it's time to captue these metrics!

In the "Step 2 - Load Sales" Sequence Container, rename the "Record Row Count" Execute SQL Task "Record Metrics". Double-click it to open the editor. Click the ellipsis on the SQLStatement property and add the following script to the existing statement:

exec ssis.usp_RecordRowSum ?,?,?,'I','Sales.SalesOrderDetail.LineTotal'

exec ssis.usp_RecordRowSum ?,?,?,'N','Sales.SalesOrderDetail.LineTotal'

exec ssis.usp_RecordRowSum ?,?,?,'C','Sales.SalesOrderDetail.LineTotal'

exec ssis.usp_RecordRowSum ?,?,?,'U','Sales.SalesOrderDetail.LineTotal'

Click the Parameter Mapping page and add twelve parameters. With the existing twelve parameters, the new twelve are numbered 12 - 23:

The parameters follow the pattern iPackageLoadID (Input, Long, Incrementally Numbered), iTaskLoadID (Input, Long, Incrementally Numbered), iSales___Amount (Input, Double, Incrementally Numbered).

Click the Ok button to close the editor.

Testing, One, Two, Three...

Execute the package to test the Summing functionality. The following represents a better report query for our collected data:

use SSiSRunTimeMetrics
go

select
m.packageName
,m.packageStartDateTime
,DateDiff(ss, m.packageStartDateTime, m.packageEndDateTime) as 'packageRunTime'
,m.packageStatus
,t.SourceName
,t.TaskStartDateTime
,DateDiff(ss, t.TaskStartDateTime, t.TaskEndDateTime) as 'taskRunTime'
,t.TaskStatus
,s.RowSum as 'Measurement'
,'Sum' as 'MeasurementType'
,st.RowTypeName
from ssis.TaskMetrics t
inner join ssis.RunTimeMetrics m on t.RunTimeMetricsID = m.id
inner join ssis.RowSums s on s.TaskMetricsID = t.TaskMetricsID
inner join ssis.RowTypes st on st.RowTypeID = s.RowSumTypeID
where m.id = (select Max(id)
from ssis.RunTimeMetrics)
and s.RowSum > 0

union

select
m.packageName
,m.packageStartDateTime
,DateDiff(ss, m.packageStartDateTime, m.packageEndDateTime) as 'packageRunTime'
,m.packageStatus
,t.SourceName
,t.TaskStartDateTime
,DateDiff(ss, t.TaskStartDateTime, t.TaskEndDateTime) as 'taskRunTime'
,t.TaskStatus
,c.[RowCount] as 'Measurement'
,'Counts' as 'MeasurementType'
,ct.RowTypeName
from ssis.TaskMetrics t
inner join ssis.RunTimeMetrics m on t.RunTimeMetricsID = m.id
inner join ssis.RowCounts c on c.TaskMetricsID = t.TaskMetricsID
inner join ssis.RowTypes ct on ct.RowTypeID = c.RowCountTypeID
where m.id = (select Max(id)
from ssis.RunTimeMetrics)
and c.[RowCount] > 0

Conclusion

Again, these examples are intended to demonstrate the principles and characteristics of ETL Instrumentation. They are not complete and Production-ready. I make no claims that this is "the right way" or even a best practice to capture ETL Run Time Metrics data. I do maintain that such data is useful in many ways - especially for troubleshooting and certain performance predictive analytics.

Next: Validation - putting this data to work.

:{> Andy

SSIS Design Pattern - ETL Instrumentation, Part 3

andyleonard — Mon, 19 Nov 2007 03:45:00 GMT

Introduction

This post is part of a series of posts on ETL Instrumentation.

In Part 1 we built a database to hold collected SSIS run time metrics and an SSIS package to deomnstrate how and why we would load metrics into the database.

In Part 2 we expanded on our database and the SSIS package to annotate version metadata, manage error metrics capture, and task status reporting.

In Part 3, we start using the ETL Instrumentation infrastructure we've built to measure some actual ETL. We now have the concepts - it's time to put them to work!

A Brief History Of Our ETL Instrumentation Project

To review, our existing database is named SSISRunTimeMetrics. It contains a schema named ssis. In this schema are eight objects:
- a table named ssis.RunTimeMetrics.
- a table named ssis.RunTimeErrors.
- a table named ssis.TaskMetrics.
- a stored procedure named ssis.usp_RecordPackageStart.
- a stored procedure named ssis.usp_RecordPackageEnd.
- a stored procedure named ssis.usp_RecordPackageError.
- a stored procedure named ssis.usp_RecordTaskStart.
- a stored procedure named ssis.usp_RecordTaskEnd.

When the task completes, we call ssis.usp_RecordTaskEnd from an Execute SQL Task. We pass in the Task Load ID from the iTaskLoadID variable, the current date and time, and the status "Succeeded".

On error, we capture Exception data and record an Error Status - both are crucial to knowing what happens when an exception is thrown.

Version Metadata

Let's create a target database for our ETL process. Open SQL Server Management Studio (SSMS) and execute the following script to create SSISRunTimeMetrics_Target:

-- create a target database...
use master
go
if not exists(select name
from master.sys.databases
where name = 'SSISRunTimeMetrics_Target')
Create Database SSISRunTimeMetrics_Target
go

Open the existing SSISRunTimeMetrics SSIS solution. Open the SSISRunTimeMetrics package and edit the version metadata:

This is a major revision - we're moving to version 2.0 and resetting the VersionBuild property. I also add text to the VersionComments property.

Don't forget the package annotation:

Now we're ready to make some code changes.

Extract, Transform, Load? Two Out Of Three Ain't Bad...

Rename the "Step 1 - Do some stuff" Sequence Container to "Step 1 - Load Contacts". Double-click the "Log Start of Task" Execute SQL Task to open the editor. Set the SQLStatement property to:

declare @Now datetime
set @Now = GetDate()
exec ssis.usp_RecordTaskStart ?,NULL,@Now,?,'Started'

Change the name property to "Load Contact":

Click the Parameter Mapping page and click the Add button. Set the Variable Name to System::TaskName, change the Data Type to VarChar, and the Parameter Name to 1.

Other applicable properties were modified in Part 2 of this series. Because of the design of the related Task Status collection in Part 2, we do not need to make major changes to the "Log Successful End of Task" or "Log Failed End of Task" Execute SQL Tasks - they are ready for our purposes. Next, delete the Test Script Task. Drag a Data Flow Task into the space vacated by the script task:

Double-click the Data Flow Task to open the editor. Drag an OLE DB Source Adapter onto the Data Flow Task canvas and double-click it to open the editor. Click the New button beside the OLE DB connection manager to open the Configure OLE DB Connection Manager dialog. Click the New button to open the Connection Manager dialog. Enter "(local)" in the Server name dropdown box. In the "Select or enter a database name" dropdown, select AdventureWorks.

You can download SQL Server 2005 Samples and Sample Databases if you don't already have them installed.

Click the OK button to close the Connection Manager dialog. Click the OK button again to close the Configure OLE DB Connection Manager dialog. Select "Person.Contact" from the "Name of the table or the view" dropdown and click the OK button to close the OLE DB Source Editor.

Drag an OLE DB Destination Adapter onto the Data Flow canvas and connect it to the OLE DB Source Adapter.

Double-click the OLE DB Destination Adapter to open the editor. Using the procedure outlined above, create a connection to the SSISRunTimeMetrics_Target database you created earlier:

Click the New button beside the "Name of the table or the view" dropdown. Change the table name in the Create Table statement from "[OLE DB Destination]" to "[dbo].[Contact]". Your script should now read:

CREATE TABLE [dbo].[Contact] (
[ContactID] INTEGER,
[NameStyle] BIT,
[Title] NVARCHAR(8),
[FirstName] NVARCHAR(50),
[MiddleName] NVARCHAR(50),
[LastName] NVARCHAR(50),
[Suffix] NVARCHAR(10),
[EmailAddress] NVARCHAR(50),
[EmailPromotion] INTEGER,
[Phone] NVARCHAR(25),
[PasswordHash] VARCHAR(40),
[PasswordSalt] VARCHAR(10),
[AdditionalContactInfo] NTEXT,
[rowguid] UNIQUEIDENTIFIER,
[ModifiedDate] DATETIME
)

Click the OK button. Click the Mappings page and then the Ok button to close the editor.

Now we have a process that extracts data from the AdventureWorks database and loads it into our SSISRunTimeMetrics_Target database. Execute the package and observe the Data Flow:

Let's modify the report query from Part 2:

select
m.packageName
,m.packageStartDateTime
,DateDiff(ss, m.packageStartDateTime, m.packageEndDateTime) as 'packageRunTime'
,m.packageStatus
,t.SourceName
,t.TaskStartDateTime
,DateDiff(ss, t.TaskStartDateTime, t.TaskEndDateTime) as 'taskRunTime'
,t.TaskStatus
,e.ErrorNumber
,e.ErrorDescription
from ssis.RunTimeMetrics m
inner join ssis.TaskMetrics t on t.RunTimeMetricsID = m.id
left outer join ssis.RunTimeErrors e on e.RunTimeMetricsID = m.id
where m.id = (select Max(id)
from ssis.RunTimeMetrics)

The left outer join to the RunTimeErrors table allows us to view successful executions. Chnaging the Where clause to bring back the latest id value from the RunTimeMetrics table will prove useful when we add tasks.

Count Me In

On the Data Flow Task, delete the data flow path between the OLE DB Source Adapter and OLE DB Destination Adapter. Drag a Row Count Tranformation from the Toolbox onto the Data Flow canvas and connect it between the OLE DB Source and OLE DB Destination Adapters.

Click SSIS, Variables and create a new package-scoped Int32 variable named iContactCount.

Double-click the Row Count transformation to open the editor. Click the VariableName dropdown and select the User::iContactCount variable to hold the record counts.

Now we have a variable to count the number of rows loaded by the data flow - we now need a place to store this data. Let's create a table to hold the row counts. Use the following script to create the ssis.RowCounts table:

As always, let's create a stored procedure to handle the insert:

use SSISRunTimeMetrics
go
if exists(select s.name + '.' + p.name
          from sys.procedures p
          inner join sys.schemas s on s.schema_id = p.schema_id
          where p.name = 'usp_RecordRowCount'
           and s.name = 'ssis')
begin
Drop Procedure ssis.usp_RecordRowCount
end
go

Create Procedure ssis.usp_RecordRowCount
@RunTimeMetricsID int
,@TaskMetricsID int
,@RowCount int
,@ParentTaskMetricsID int = null
As
begin
-- insert the run time errors data...
insert into ssis.RowCounts
(TaskMetricsID
,RunTimeMetricsId
,ParentTaskMetricsID
,[RowCount])
values
(@TaskMetricsID
,@RunTimeMetricsID
,@ParentTaskMetricsID
,@RowCount)
end
go

There are different ways (methodologies) to instrument an ETL SSIS package. Some designers choose to place the code to insert row counts - as well as the code to store start and end metrics we capture in the ssis.TaskMetrics table - on the Control Flow. Other choose to use Event Handlers - OnPreExecute and OnPostExecute - to house the logic performing these operations.

There are sound reasons for choosing each method. An argument for using Event Handlers is it provides a separation of the ETL logic and metrics / status logic. An argument for not using Event Handlers is all the logic is readily accessible on the Control Flow and Data Flow. Both arguments are legitimate and the customer's maintenance needs and SSIS experience should be considered in this design decision.

For the purposes of this series, I choose to have most of the logic visible on the Control Flow and Data Flow.

Return to the Control Flow to add a mechanism to insert the RowCount data into our new table. Drag an Execute SQL Task into the "Step 1 - Load Contact" Sequence Container. Connect a Success Precendence Constraint from the "Log Successful End of Task" Execute SQL Task to the new Execute SQL Task.

Double-click the new Execute SQL Task to open the editor. Configure the General and Parameter Mappings pages as shown:

Click the OK button to close the Execute SQL Task editor.

Before we test the new row count functionality, let's clear the destination table. Execute the following SQL statement in SSMS:

Truncate Table SSISRunTimeMetrics_Target.dbo.Contact

Now execute the package.

A slight modification of our earlier report query adds the record counts:

select
m.packageName
,m.packageStartDateTime
,DateDiff(ss, m.packageStartDateTime, m.packageEndDateTime) as 'packageRunTime'
,m.packageStatus
,t.SourceName
,t.TaskStartDateTime
,DateDiff(ss, t.TaskStartDateTime, t.TaskEndDateTime) as 'taskRunTime'
,t.TaskStatus
,c.[RowCount]
,e.ErrorNumber
,e.ErrorDescription
from ssis.RunTimeMetrics m
inner join ssis.TaskMetrics t on t.RunTimeMetricsID = m.id
left outer join ssis.RunTimeErrors e on e.RunTimeMetricsID = m.id
left outer join ssis.RowCounts c on c.RunTimeMetricsID = m.id
where m.id = (select Max(id)
from ssis.RunTimeMetrics)

The results appear as shown:

We get all the rows this execution - cool.

This demostration is nearly true-to-life. Let's take it closer by adding incremental loading. (I cover another SSIS Design Pattern - Incremental Loads - in another post.)

First return to the Control Flow. Click anywhere in the Control Flow canvas where there is white space. This sets focus to the Control Flow, which sets focus to the Package level. Click SSIS, Variables to display the Variables dialog. Add three package-scoped Int32 variables named iNewRowsCount, iChangedRowsCount, and iUnchangedRowsCount.

Click the Data Flow tab and drag an OLE DB Command, Lookup, and Conditional Split transformations onto the Data Flow Task. Also drag three Row Count transformations onto the Data Flow.

Rename the OLE DB Destination Adapter "Insert New Rows" and delete the data flow path (the green line / arrow) that connects it to the Row Count transformation. Connect the Lookup to the output of the Row Count to the input of the Lookup and double-click the Lookup to open the editor. Set the OLE DB connection manager property to "(local).SSISRunTimeMetrics_Target" and the "Use a table or view" dropdown to "[dbo].[Contact]":

Click the Columns tab. Right-click in the mappings area and click "Select All Mappings". Right-click in the mappings area again and click "Delete Selected Mappings". Drag ContactID from the Available Input Columns and drop it on the Available Lookup Columns' ContactID column.

Select all the columns in the Available Lookup Columns list except the AdditionalContactInfo and rowguid columns. In the Output Alias column, add a "Dest_" prefix to each field. This will allow easier identification in the pipeline:

Click the Configure Error Output button and change the Lookup Output Error dropdown from "Fail component" to "Ignore failure". This changes the default behavior of the Loopup operation from an Inner Join to a Left Outer Join.

Click the OK button to close the Lookup editor. Connect the output of the Lookup transformation to the input of the Conditional Split transformation and double-click the Conditional Split to open the editor.

Name the first Output "New Rows" and set the Condition to "IsNull(Dest_ContactID)".

Name the second Output "Changed Rows" and set the Condition to "(IsNull(Dest_NameStyle) ? True : Dest_NameStyle) != (IsNull(NameStyle) ? True : NameStyle) || (IsNull(Dest_Title) ? "null" : Dest_Title) != (IsNull(Title) ? "null" : Title) || (IsNull(Dest_FirstName) ? "null" : Dest_FirstName) != (IsNull(FirstName) ? "null" : FirstName) || (IsNull(Dest_MiddleName) ? "null" : Dest_MiddleName) != (IsNull(MiddleName) ? "null" : MiddleName) || (IsNull(Dest_LastName) ? "null" : Dest_LastName) != (IsNull(LastName) ? "null" : LastName) || (IsNull(Dest_Suffix) ? "null" : Dest_Suffix) != (IsNull(Suffix) ? "null" : Suffix) || (IsNull(Dest_EmailAddress) ? "null" : Dest_EmailAddress) != (IsNull(EmailAddress) ? "null" : EmailAddress) || (IsNull(Dest_EmailPromotion) ? -1 : Dest_EmailPromotion) != (IsNull(EmailPromotion) ? -1 : EmailPromotion) || (IsNull(Dest_Phone) ? "null" : Dest_Phone) != (IsNull(Phone) ? "null" : Phone) || (IsNull(Dest_PasswordHash) ? "null" : Dest_PasswordHash) != (IsNull(PasswordHash) ? "null" : PasswordHash) || (IsNull(Dest_PasswordSalt) ? "null" : Dest_PasswordSalt) != (IsNull(PasswordSalt) ? "null" : PasswordSalt) || (IsNull(Dest_ModifiedDate) ? (DT_DBDATE)"1/1/1901" : Dest_ModifiedDate) != (IsNull(ModifiedDate) ? (DT_DBDATE)"1/1/1901" : ModifiedDate)".

Change the Default Output Name to "Unchanged Rows".

Drag an output from the Conditional Split to one of the Row Count transformations. When prompted, select the New Rows output from the Conditional Split transformation.

Rename this Row Count "New Rows Count" and double-click it to open the editor. Set the VariableName property to User::iNewRowsCount and click the OK button to close the editor.

Connect the output of the New Rows Count transformation to the OLE DB Destination Adapter "Insert New Rows".

Drag a second output from the Conditional Split to another Row Count transformation. When prompted, select the Changed Rows output.

Rename the Row Count transformation "Changed Rows Count" and double-click the Row Count transformation to open the editor. Set the VariableName property to iChangedRowsCount and click the OK button to close the editor.

Double-click the OLE DB Command transformation to open the advanced editor. Set the Connection Manager property to "(local).SSISRunTimeMetrics_Target".

Click the Component Properties tab. Click the SQLCommand ellipsis and enter the following statement in the String Value Editor:

update dbo.Contacts
set NameStyle = ?
,Title = ?
,FirstName = ?
,MiddleName = ?
,LastName = ?
,Suffix = ?
,EmailAddress = ?
,EmailPromotion = ?
,Phone = ?
,PasswordHash = ?
,PasswordSalt = ?
,ModifiedDate = ?
where ContactID = ?

Connect the remaining Conditional Split output (Unchanged Rows) to the remaining Row Count transformation. Rename the Row Count "Unchanged Rows Count", double-click it to open the editor, and set the VariableName to User::iUnchangedRowCount.

Some Optimization...

Before we leave the Data Flow, let's optimize it for performance. There's a couple good reasons for this, one big reason is the current design may very well result in blocking, maybe even deadlocks!

Delete the "Update Changed Rows" OLE DB Command. Drag an OLE DB Destination Adapter onto the Data Flow Task and connect it to the Changed Rows Count transformation. Change the name of the Adapter to "stgContactChangedRows" and double-click it to open the editor. Set the OLE DB connection manager to "(local).SSISRUnTimeMetrics_Target". Change the Data Access Mode to "Table or view". Click the New button beside the "Name of table or view" dropdown:

Click the OK button to create the new table, then click the Mappings page to complete configuration. Click the OK buton to close the editor.

On the Control Flow, add an Execute SQL Task just after the Data Flow Task. Name it "Apply Staged Updates" and set the Connection property to "(local).SSISRunTimeMetrics_Target". Set the SQLStatement property to the follow SQL:

UPDATE c
SET c.ContactID = u.ContactID
,c.NameStyle = u.NameStyle
,c.Title = u.Title
,c.FirstName = u.FirstName
,c.MiddleName = u.MiddleName
,c.LastName = u.LastName
,c.Suffix = u.Suffix
,c.EmailAddress = u.EmailAddress
,c.EmailPromotion = u.EmailPromotion
,c.Phone = u.Phone
,c.PasswordHash = u.PasswordHash
,c.PasswordSalt = u.PasswordSalt
,c.AdditionalContactInfo = u.AdditionalContactInfo
,c.rowguid = u.rowguid
,c.ModifiedDate = u.ModifiedDate
FROM dbo.Contact c
INNER JOIN dbo.stgContactChangedRows u on u.ContactID = c.ContactID

This converts the RBAR (Row-By-Agonizing-Row) update statement in the OLE DB Command to a set-based update. Much better.

We need to capture these row counts, but we need a way to mark them as well. Let's return to our ssis.RowCounts table design and modify it. Before modifying the table, let's create and populate a lookup table for RowCountTypes:

Execute the following statement to add the lookup column to the ssis.RowCounts table:

if not exists(select c.name
from sys.columns c
inner join sys.tables t on t.object_id = c.object_id
inner join sys.schemas s on s.schema_id = t.schema_id
where s.name = 'ssis'
and t.name = 'RowCounts'
and c.name = 'RowCountTypeID')
Alter Table ssis.RowCounts Add RowCountTypeID char(1) null
go

Execute the following statements to populate the RowCountTypes table:

if not exists(select RowCountTypeID
from ssis.RowCountTypes
where RowCountTypeID = 'I')
insert into ssis.RowCountTypes
(RowCountTypeID
,RowCountTypeName
,RowCountTypeDescription)
values
('I'
,'Selected Input Rows'
,'Input rows selected from a source')

if not exists(select RowCountTypeID
from ssis.RowCountTypes
where RowCountTypeID = 'N')
insert into ssis.RowCountTypes
(RowCountTypeID
,RowCountTypeName
,RowCountTypeDescription)
values
('N'
,'New Rows'
,'New rows')
if not exists(select RowCountTypeID
from ssis.RowCountTypes
where RowCountTypeID = 'C')

insert into ssis.RowCountTypes
(RowCountTypeID
,RowCountTypeName
,RowCountTypeDescription)
values
('C'
,'Changed Rows'
,'Changed rows')
if not exists(select RowCountTypeID
from ssis.RowCountTypes
where RowCountTypeID = 'U')

insert into ssis.RowCountTypes
(RowCountTypeID
,RowCountTypeName
,RowCountTypeDescription)
values
('U'
,'Unchanged Rows'
,'No changes detected in rows')

We now need to updated our stored procedure that inserts records into ssis.RowCounts:

use SSISRunTimeMetrics
go

if exists(select s.name + '.' + p.name
          from sys.procedures p
          inner join sys.schemas s on s.schema_id = p.schema_id
          where p.name = 'usp_RecordRowCount'
           and s.name = 'ssis')
begin
Drop Procedure ssis.usp_RecordRowCount
end
go

Create Procedure ssis.usp_RecordRowCount
@RunTimeMetricsID int
,@TaskMetricsID int
,@RowCount int
,@RowCountType char(1)
,@ParentTaskMetricsID int = null
As
begin
-- insert the run time errors data...
insert into ssis.RowCounts
(TaskMetricsID
,RunTimeMetricsId
,ParentTaskMetricsID
,[RowCount]
,RowCountTypeID)
values
(@TaskMetricsID
,@RunTimeMetricsID
,@ParentTaskMetricsID
,@RowCount
,@RowCountType)
end
go

Finally, let's update the "Record Row Count" SQLStatement.While we're in there, let's add the inserts for the additional row count variables. Return to the Control Flow and double-click the "Record Row Count" Execute SQL Task to open the editor.

Change the SQLStatement to:

exec ssis.usp_RecordRowCount ?,?,?,'I'
exec ssis.usp_RecordRowCount ?,?,?,'N'
exec ssis.usp_RecordRowCount ?,?,?,'C'
exec ssis.usp_RecordRowCount ?,?,?,'U'

Click the Parameter Mappings page and edit as shown:

Note every third parameter is a count measurement, the order indicated by the SQLStatement property. Click the OK button to close the editor and execute the package.

Let's modify our report query yet again to have a look at the collected metrics:

select
m.packageName
,m.packageStartDateTime
,DateDiff(ss, m.packageStartDateTime, m.packageEndDateTime) as 'packageRunTime'
,m.packageStatus
,t.SourceName
,t.TaskStartDateTime
,DateDiff(ss, t.TaskStartDateTime, t.TaskEndDateTime) as 'taskRunTime'
,t.TaskStatus
,c.[RowCount]
,ct.RowCountTypeName
,e.ErrorNumber
,e.ErrorDescription
from ssis.RunTimeMetrics m
inner join ssis.TaskMetrics t on t.RunTimeMetricsID = m.id
left outer join ssis.RunTimeErrors e on e.RunTimeMetricsID = m.id
left outer join ssis.RowCounts c on c.RunTimeMetricsID = m.id
left outer join ssis.RowCountTypes ct on ct.RowCountTypeID = c.RowCountTypeID
where m.id = (select Max(id)
from ssis.RunTimeMetrics)

The results?

The Data It Is A-Changin'

Let's simulate live data changes by making some changes in the source data. Execute the following statements to delete and modify data in SSISRunTimeMetrics_Target:

use SSISRunTimeMetrics_Target

delete dbo.Contact
where ContactID % 17 = 0

update dbo.Contact
set MiddleName = IsNull(MiddleName,'') + 'A'
where ContactID % 25 = 0

Execute the package again and observe the Data Flow. Now this is starting to look like SSIS ETL!

Let's have a look at our metrics to see if the numbers match:

I'd say that settles it.

We are now capturing enough information to gain good insight into our ETL process - how cool is that? More next time...

Get the code!

:{> Andy

SSIS Design Pattern - ETL Instrumentation, Part 2

andyleonard — Sun, 11 Nov 2007 06:18:00 GMT

Introduction

This post is part of a series of posts on ETL Instrumentation.

In Part 1 we built a database to hold collected SSIS run time metrics and an SSIS package to deomnstrate how and why we would load metrics into the database.

In Part 2 we will expand on our database and the SSIS package to annotate version metadata, manage error metrics capture, and task status reporting.

A Brief History Of Our ETL Instrumentation Project

To review, our existing database is named SSISRunTimeMetrics. It contains a schema named ssis. In this schema are three objects:
- a table named ssis.RunTimeMetrics.
- a stored procedure named ssis.usp_RecordPackageStart.
- a stored procedure named ssis.usp_RecordPackageEnd.

First Things First

Let's start by opening the SSIS package we built in Part 1. The first thing we want to change is the package metadata - the data in our SSIS package about our SSIS package. Why? We always want to know what changes were made, who made the changes, and when they were made.

Let's begin by updating the version. Click on the Control Flow tab and press the F4 key to open properties. Scroll to the Version Properties and change the VersionMinor from 0 to 1:

Also edit the Control Flow annotation that contains version information:

Update package description annotation. I like to include a Change Log summary inside each package:

In the Change Log, I include

A version number in the format Major.Minor.Date.
The date of the change.
The name of the developer who made the change.
A brief description of the change.

Now I'm ready to actually make changes! To facilitate package execution on your machine, I want to edit the name and target SQL Server instance of the OLEDB Connection Manager. Double-click the connection manager to edit it and change the Server name property to "(local)".

Click the OK button to close the connection manager editor. For consistency, right-click the connection manager and click Rename. Change the name to "(local).SSISRunTimeMetrics":

There. Housekeeping accomplished!

A test run verifies I haven't broken anything:

Unit test successful - ready to continue developing.

Errors Aside

Our current solution works well as long as our SSIS package succeeds. And this is the way all good engineering is done; Use Case 1 should always be "Everything Succeeds". But what happens if the SSIS package fails? Shouldn't we record the failure and as much metadata about the failure as possible? I'm glad you asked: Yes, we should.

In SSIS, errors "bubble". The concept of bubbling is familiar to ASP developers - it simply means events follow a prescribed path through the scope hierarchy of the application. Ok, you can scratch the word "simply" out of the preceeding sentence.

Let's use the script task to generate an error. Right the Test Script Task and click Edit to open the editor. Click the Script page and the Design Script button to open the Visual Studio for Applications (VSA) script editor. I use Script Tasks for exception testing. The following code - replacing the "Dts.TaskResult = Dts.Results.Success" line - will prompt you to have the task succeed or fail:

Dim iResponse As Integer
iResponse = MsgBox("Have task complete with Success?", MsgBoxStyle.YesNo)
If iResponse = vbYes Then
Dts.TaskResult = Dts.Results.Success
Else
Dts.TaskResult = Dts.Results.Failure
End If

The additional code causes a second messagebox:

If you click Yes in response, the task will succeed. Clicking No will cause the task to fail:

SSIS has some great logging and error-handling functionality built in. The first place to look when an error occurs is the Progress / Execution Results tab in the SSIS Designer GUI. The tab is named Progress while the package is executing and becomes Execution Results once execution completes. Click the Progress / Execution Results tab to view the error raised when you clicked the No button:

This isn't a very descriptive error message. Some SSIS error messages aren't very descriptive. Let's add some error logging capabilities to our ETL Instrumentation application.

To Err is Human...

There are lots of ways to develop solutions with SSIS. Some of them are best practices, more are simply matters of coding style. For instance, here you could build the database table to hold error information first; or you could build the error-handling functionality in the SSIS package first. This is a style decision. I choose to build the table first.

My initial errors table will be named RunTimeErrors and will contain the following fields:

RunTimeMetricsID - to map the error to the execution instance. I want to know the error occurred when I executed the package at a certain time.
ErrorDateTime - when the error actually occurred.
SourceName - the task that raised the error.
ErrorNumber - the error number.
ErrorDescription - the error description.

I generate the table with the following T-SQL script:

Note this script is re-runnable. I encourage you to take the time to write re-runnable scripts when developing database solutions.

We could stop here and directly insert rows into our error table, but let's complete the development process by creating a stored procedure to do this work for us. Execute the following code in SSMS:

use SSISRunTimeMetrics
go

if exists(select s.name + '.' + p.name
          from sys.procedures p
          inner join sys.schemas s on s.schema_id = p.schema_id
          where p.name = 'usp_RecordPackageError'
           and s.name = 'ssis')
begin
Drop Procedure ssis.usp_RecordPackageError
end
go

Create Procedure ssis.usp_RecordPackageError
@RunTimeMetricsID int
,@SourceName varchar(255) = null
,@ErrorNumber int = null
,@ErrorDescription varchar(2500) = null
As

begin

  -- set @Now...
declare @Now datetime
set @Now = GetDate()

-- insert the run time errors data...
insert into ssis.RunTimeErrors
(RunTimeMetricsID
,ErrorDateTime
,SourceName
,ErrorNumber
,ErrorDescription)
values
(@RunTimeMetricsID
,@Now
,@SourceName
,@ErrorNumber
,@ErrorDescription)

end

go

Again, re-runnable code. Now lets' return to Visual Studio to populate our table with error information.

Click on the Test Script Task on the Control Flow. Then click the Event Handlers tab. Because you clicked the Test Script Task first, the Event Handler Executable defaults to the Test Script Task. The Event Handler defaults to the OnError event.

To create the OnError Event Handler, click the link labeled "Click here to create an 'OnError' event handler for executable 'Test Script Task'". Let's do that now.

After you click the link you will note the canvas looks remarkably familiar: it's nearly identical to the Control Flow.

Our goal is to load data into our RunTimeErrors table. Let's use an Execute SQL Task to call the stored procedure. Drag an Execute SQL Task onto the OnError Event Handler canvas and double-click it to open the editor. Name the Execute SQL Task "Record Error". Set the connection to "(local).SSISRunTimeMetrics". Set the SQLStatement property to "exec ssis.usp_RecordPackageError ?,?,?,?":

Click the Parameter Mapping page. Parameterized queries use question marks (?) as parameter placeholders. There are a few ways to access the parameters (a subject for another blog post). We will use ordinal position to access them. Click the Add button to add a new parameter. Set the Variable Name property to "User::iPackageLoadID" and the Parameter Name property to "0". The default data type ("Long") works for SSIS integer data type variables. This maps the iPackageLoadID variable to the first question mark in our SQLStatement property, which is the @RunTimeMetricsID parameter for the stroed procedure. Got it? Good.

Add three more parameters:

Variable Name: System::SourceName
Data Type: VarChar
Parameter Name: 1

Variable Name: System::ErrorCode
Data Type: Long
Parameter Name: 2

Variable Name: System::ErrorDescription
Data Type: VarChar
Parameter Name: 3

Click the OK button to close the editor and execute the package.

Click the OK button on the first messagebox, then click the No button on the messagebox dialog to fail the Test Script Task. When the task fails, the OnError event fires and calls our stored procedure:

Let's look at the table to see what was recorded. Return to SSMS and execute the following query:

select *
from ssis.RunTimeErrors
order by ErrorDateTime desc

The results:

We now have basic error information from our failed task. This is great for our Test Script Task, but what about the other tasks on our SSIS package? Do we have to create OnError event handlers for them as well? That could be a lot of work.

Remember that bubbling thing we mentioned at the beginning of this post? We can employ the fact that errors bubble up through the object model. When an error occurs in the Test Script Task, an error bubbles up to the Sequence Container named "Step 1 - Do some stuff". You can see it in this earlier screenshot:

To demonstrate, let's add error handling to the Sequence Container. First, stop debugging the SSIS package if you have not already done so. Then right-click the "Record Error" Execute SQL Task and click Copy to copy it to the clipboard. Click the Executable dropdown and select the "Step 1 - Do some stuff" Sequence Container:

Select the OnError event handler and click the link to create the OnError event handler. Then paste the "Record Error" Execute SQL Task onto the new OnError event handler canvas. Re-execute the package observing the new OnError event handler. Click the No button to cause the Test Script Task to fail.

Note the new OnError event handler fires, in addition to the Test Script Task OnError event handler:

Re-execute the query in SSMS to verify:

Note the subsequent execution (RunTimeMetricsID 10, in my case) records the error twice. Hmmm... this is not the desired behavior. I only want the error to show up once per instance. However, also note the information recorded is nearly identical - only the date is different (and then only by 183 milliseconds). So the 20:17:58.803 occurrence of the error is from the Test Script Task and the 20:17:58.990 occurrence is the same error bubbling into the "Step 1 - Do some stuff" Sequence Container's context and firing an identicaly-configured OnError event handler.

Again, this is not the desired behavior but it gives me an idea: Since I know errors bubble up the object model, all errors should eventually make it to the Package level. If I delete my existing OnError event handlers and create a single OnError event handler at the package level, I should then be able to record any and all errors - no matter where they occur in the SSIS package.

Let's do that. First, copy the "Record Error" Execute SQL Task to the clipboard. Click the Delete button to delete the Sequence Container's OnError event handler, then navigate to the Test Script Task's OnError event handler and click the Delete button there. Click the Yes button on the confirmation dialog each time:

Navigate to the SSISRunTimeMetrics (package) event handlers:

Create the event handler by clicking the link and paste the Execute SQL Task onto the canvas. Voila, you now have error handling for then entire package. Here we must also add package status logging. The quickest way to accomplish this is to return to the Control Flow and copy the "Log End of Package Execution" Execute SQL Task to the clipboard. Navigate to the package OnError event handler and paste the Execute SQL Task onto the canvas. Connect it to the "Record Error" script task as shown:

Double-click the task to open the editor. Rename the task "Log Failed Package Execution" and change the SQLStatement property to:

declare @Now datetime
set @Now = GetDate()
Exec ssis.usp_RecordPackageEnd ?,@Now,'Failed'

Click the OK button to close the editor.

Re-execute the package to test, answering No to the dialog and failing the Script Task. Re-execute the query in SSMS to verify a single instance of the error is recorded:

Note the SourceName column contains the name of the task where the error originates, even though the error has bubbled through two layers of object model. This also holds for the ErrorNumber and ErrorDescription fields.

Task Status, Anyone?

It's helpful to be able to track performance at a granular level. When instrumenting an ETL process, I like to record additional (and different) metadata at the Task level.

Since containers contain tasks and groups can represent linked processes or related tasks, task metrics must include parent-task relationships. Execute the following script to build our first TaskMetrics table:

As before, let's create stored procedures to manage inserting start-of-task information into the TaskMetrics table and updating end-of-task information:

use SSISRunTimeMetrics
go

if exists(select s.name + '.' + p.name
          from sys.procedures p
          inner join sys.schemas s on s.schema_id = p.schema_id
          where p.name = 'usp_RecordTaskMetrics'
           and s.name = 'ssis')
begin
Drop Procedure ssis.usp_RecordTaskMetrics
end
go

Create Procedure ssis.usp_RecordTaskStart
@RunTimeMetricsID int
,@ParentTaskMetricsID int = null
,@TaskStartDateTime datetime = null
,@SourceName varchar(255) = null
,@TaskStatus varchar(25) = null
As

begin

-- set @TaskStartDateTime default...
declare @Now datetime
set @Now = GetDate()
if @TaskStartDateTime is null
set @TaskStartDateTime = @Now

-- insert the run time errors data...
insert into ssis.TaskMetrics
(RunTimeMetricsID
,ParentTaskMetricsID
,TaskStartDateTime
,SourceName
,TaskStatus)
output inserted.TaskMetricsID as 'Inserted_Task_ID'
values
(@RunTimeMetricsID
,@ParentTaskMetricsID
,@TaskStartDateTime
,@SourceName
,@TaskStatus)

end

go

Now for the end-of-task update:

use SSISRunTimeMetrics
go

if exists(select s.name + '.' + p.name
          from sys.procedures p
          inner join sys.schemas s on s.schema_id = p.schema_id
          where p.name = 'usp_RecordTaskEnd'
           and s.name = 'ssis')
begin
Drop Procedure ssis.usp_RecordTaskEnd
end
go

Create Procedure ssis.usp_RecordTaskEnd
@TaskMetricsID int
,@TaskEndDateTime datetime = null
,@TaskStatus varchar(25) = null
As

begin

-- set @TaskEndDateTime default...
declare @Now datetime
set @Now = GetDate()

if (@TaskEndDateTime Is Null)
set @TaskEndDateTime = @Now

-- update the task metrics data...
update ssis.TaskMetrics
Set TaskEndDateTime = @TaskEndDateTime
   ,TaskStatus = @TaskStatus
where TaskMetricsid = @TaskMetricsid

end

go

Instrumenting the Task

We will use the new procedures to instrument the task.

We need some variables for these parameters. At a bare minimum, we need a variable to hold the Task Metrics ID returned from the ssis.usp_RecordTaskStart stored procedure. Open variables and create a new package-scoped, Int32 type variable named iTaskLoadID:

Drag an Execute SQL Task into the Sequence Container an configure the General page as shown:

Set the SQLStatement property to the follow script:

declare @Now datetime
set @Now = GetDate()
exec ssis.usp_RecordTaskStart ?,NULL,@Now,'Do some stuff','Started'

Click the Parameter Mapping page and add a parameter. Set the Variable Name to User::iPackageLoadID and Parameter Name to 0 - accept the defaults for the remaining parameter properties. Click the Result Set page and add a new result. Set the ResultName to 0 and the Variable Name to User::iTaskLoadID. This will hold the TaskMetricsID value returned from the ssis.usp_RecordTaskStart stored procedure:

Click the OK button to close the editor. Next, let's create a way to update TaskMetrics table after a successful execution.

Drag an Execute SQL Task into the "Step 1 - Do some stuff" Sequence Container and connect it as shown:

Double-click the Execute SQL Task to edit it. Name it "Log Successful End of Task". Set the SQLStatement to:

declare @Now datetime
set @Now = GetDate()
exec ssis.usp_RecordTaskEnd ?,@Now,'Completed'

Add one parameter mapped to User::iTaskLoadID and set to Parameter Name 0. Test the new tasks by executing the SSIS package. When prompted, have the script task succeed. Execute the following query to observe the results:

select *
from ssis.TaskMetrics
order by TaskMetricsID desc

Note TaskMetricsID 2 completed while TaskMetricsID 1 started and then stopped.

We now need to add a way to track and record task error status. Add another Execute SQL Task to the Control Flow canvas and connect a Failed Precedence Constraint from the Sequence Container to the new Execute SQL Task as shown:

Name the new Execute SQL Task "Log Failed End of Task". Set the SQLStatement to:

declare @Now datetime
set @Now = GetDate()
exec ssis.usp_RecordTaskEnd ?,@Now,'Failed'

Add one parameter mapped to User::iTaskLoadID and set to Parameter Name 0. Test the new tasks by executing the SSIS package. When prompted, have the script task fail. Execute the query to observe the results:

You now have enough information to correlate the three tables. For example, we can view the results of the last execution with the following script:

select
m.packageName
,m.packageStartDateTime
,DateDiff(ss, m.packageStartDateTime, m.packageEndDateTime) as 'packageRunTime'
,m.packageStatus
,t.SourceName
,t.TaskStartDateTime
,DateDiff(ss, t.TaskStartDateTime, t.TaskEndDateTime) as 'taskRunTime'
,t.TaskStatus
,e.ErrorNumber
,e.ErrorDescription
from ssis.RunTimeMetrics m
inner join ssis.RunTimeErrors e on e.RunTimeMetricsID = m.id
inner join ssis.TaskMetrics t on t.RunTimeMetricsID = m.id
where m.packageStartDateTime = (select Max(packageStartDateTime)
from ssis.RunTimeMetrics)

Those results appear as shown here:

This makes a fine source query for a SQL Server Reporting Services solution!

Conclusion

In this second part of ETL Instrumentation, we modified version metadata to indicate we have made changes; added package error logging and error status logging; and added task status logging.

Get the code!

:{> Andy

SSIS Design Pattern - ETL Instrumentation, Part 1

andyleonard — Fri, 03 Aug 2007 07:36:00 GMT

Introduction

SSIS is a fantastic ETL engine. When I build an ETL solution in SSIS, I like to collect runtime metrics. Why? I use the data initially to determine a baseline for performance and to determine, in some cases, if I'm loading within a defined window.

I refer to this process of collecting performance and runtime metrics as ETL Instrumentation and I collect this data for every enterprise ETL solution I design.

The types of things I want to know are:

Did the process start when I wanted it to start?
Did it succeed?

If not, is it still running?
Did it fail?

How long did it execute?
How does recent performance compare to past performance?

In this mini-series entitled SSIS Design Pattern - ETL Instrumentation, I will demonstrate how I capture, report, and measure these data using SSIS, SSRS, SQL Agent Jobs, and Transact-SQL.

In Part 1, we build an elementary system. In subsequent parts, we will improve upon the concepts demonstrated here.

To Instrument Or Not To Instrument; That Is The Question

My thoughts on this are best summed up thus:

There are two types of enterprise data warehouse developers:

1) Those who collect ETL instrumentation data; and

2) Those who will.

Why do I make this statement? When you build ETL to load your enterprise data warehouse, you are building the source for the reporting system. Should anything unfortunate happen to the data between the time you read it from the System Of Record and the time it lands in your Reporting Server Database, you are basically guilty until proven innocent. ETL Instrumentation will provide evidence for (and sometimes against) you as you defend your ETL design and implementation.

Building The Database

Before we instrument an ETL process, we need a database solution for the data. We'll start by building the database itself, then add objects.

Let's build the database. We'll call it SSISRunTimeMetrics. Open SQL Server Management Studio (SSMS) and execute the following Transact-SQL:

use master
go

if not exists(select name
from master.sys.databases
where name = 'SSISRunTimeMetrics')
Create Database SSISRunTimeMetrics
go

This script creates the SSISRunTimeMetrics database. Next, let's create a schema and table to hold our metrics:

use SSISRunTimeMetrics
go

-- vars...
declare @sql varchar(255)

-- create ssis schema...
if not exists(select name
from sys.schemas
where name = 'ssis')
begin
set @sql = 'Create Schema ssis'
exec(@sql)
end

-- create RunTimeMetrics table...
if not exists(select s.name + '.' + t.name
              from sys.tables t
              inner join sys.schemas s on s.schema_id = t.schema_id
              where t.name = 'RunTimeMetrics'
               and s.name = 'ssis')
Create Table ssis.RunTimeMetrics
(id int identity(1,1)
,packageStartDateTime datetime null
,packageEndDateTime datetime null
,packageName varchar(255) null
,packageVersion varchar(25) null
,packageStatus varchar(25) null)

Next, let's add a stored procedure to record a package start event:

use SSISRunTimeMetrics
go

if exists(select s.name + '.' + p.name
          from sys.procedures p
          inner join sys.schemas s on s.schema_id = p.schema_id
          where p.name = 'usp_RecordPackageStart'
           and s.name = 'ssis')
begin
Drop Procedure ssis.usp_RecordPackageStart
end
go

Create Procedure ssis.usp_RecordPackageStart
@packageStartDateTime datetime = null
,@packageName varchar(255) = null
,@packageVersion varchar(25) = null
,@packageStatus varchar(25) = null
As
begin
-- set @packageStartDateTime default...
declare @Now datetime
set @Now = GetDate()

if (@packageStartDateTime Is Null)
   set @packageStartDateTime = @Now

-- insert the run time metrics data...
insert into ssis.RunTimeMetrics
   (packageStartDateTime
   ,packageName
   ,packageVersion
   ,packageStatus)
values
(@packageStartDateTime
,@packageName
,@packageVersion
,@packageStatus)

end

go

That's all for now. We'll add more stuff later.

Building The SSIS Project

Open SQL Server Business Intelligence Development Studio (BIDS) and create a new project. When the New Project dialog displays, select Business Intelligence Projects from the Project types list and Integration Services Project from the Templates. Name the project SSISRunTimeMetrics and click the OK button:

Rename the Package SSISRunTimeMetrics.dtsx:

Always click the Yes button when you see this dialog (trust me):

Right-click in the white-space on the Control Flow canvas and click Add Annotation to document your code:

I can't say enough about good documentation. You never know when someone else will have to open your source code to troubleshoot.

Right-click in the white-space of the Control Flow canvas and click Variables. Add a variable named sVersion. Make sure the Scope is the SSISRunTimeMetrics package and make the Data Type String:

Add a Sequence Container and change the name to "Step 1 - Do some stuff":

Sequence Containers are nice for several reasons:

They're aesthetic. You can break up chucks of functionality logically... how cool!
When troubleshooting, you can execute the contents of a container from the right-click context menu.
You can isolate chunks of functionality using transactions by container.

Next, add an Execute SQL Task and name it "Log Start of Package Execution":

Double-click the Execute SQL Task to edit it. Click Connection and select "<New connection...>":

When the Configure OLE DB Connection Manager dialog displays, click the "New..." button:

Enter or select your server in the Server Name dropdown. In the "Connect to a database" panel, select the "Select or enter a database name" option and enter or select the SSISRunTimeMetrics database you created earlier. Click the OK button to proceed:

Then click the OK button again to set the Connection property of the Execute SQL Task:

Select the SQL Statement property and click the ellipsis. When the "Enter SQL Query" dialog displays, enter the following parameterized Transact-SQL statement:

Exec ssis.usp_RecordPackageStart ?,?,?,'Started'

Click the OK button to proceed:

Click on the Parameter Mapping page and configure parameters as shown:

Click the OK button to proceed.

Next, let's do some work on the sVersion variable. Click on the sVersion variable name and press the F4 key to display variable properties. Change the EvaluateAsExpression property to True, then click on the Expression property's ellipsis:

When the Expression Builder displays, enter the following SSIS Expression Language text to build a version string in the format Major.Minor.Build:

(DT_WSTR, 3) @VersionMajor + "." + (DT_WSTR, 3)@VersionMinor + "." + (DT_WSTR, 3)@VersionBuild

Click the "Evaluate Expression" button to check the expression. Then click the OK button to close the Expreesion Builder:

Add a Script Task and configure it to popup a message box displaying the value of sVersion:

Imports System
Imports System.Data
Imports System.Math
Imports Microsoft.SqlServer.Dts.Runtime

Public Class ScriptMain

Public Sub Main()

Dim sVersion As String = Dts.Variables("sVersion").Value.ToString
Dim sMsg As String

sMsg = "Version: " & sVersion

MsgBox(sMsg)

Dts.TaskResult = Dts.Results.Success

End Sub

End Class

Execute the package. Then execute the following Transact-SQL query to check the results:

select * from ssis.RunTimeMetrics

It's time to add a procedure to log the end of package execution, but we have a problem: How do we identify the row we just added to the ssis.RunTimeMetrics table? Think about it. We just inserted a row, and since that's the only row in the table it's pretty simple to identify. Later, we will have lots of rows in the table in varying states. We need a method to positively identify the row we're working with.

We could use the last row inserted - apply Max(id) criteria - but what happens as our metrics grows to include multiple packages? One package could start while another was executing and we'd update the wrong row.

We require a means to positively identify the row when added to the table. Such a means exists. If we modify the ssis.usp_RecordPackageStart stored procedure we can have it return the value of the inserted ID. The OUTPUT clause gives us access to the inserted and deleted virtual tables. Modify the stored procedure by adding the following OUTPUT clause as shown:

output inserted.id as 'Inserted_ID'

Execute the Transact-SQL to drop and recreate the ssis.usp_RecordPackageStart stored procedure:

Add another Int32 Type variable named iPackageLoadID:

Back in Execute SQL Task, change the ResultSet property to "Single row":

Click the Result Set page and add a new resultset. Name it 0 (ordinals work well) and assign the results of this query to the iPackageLoadID variable you just created. This will push the results of the OUTPUT clause in the stored procedure (which will include the id of the newly inserted row) into the iPackageLoadID variable. You can then access it later in the package to update the precise row:

Alter the Script Task by adding iPackageLoadID to the list of ReadOnlyVariables:

Alter the VB.Net script to read:

Imports System
Imports System.Data
Imports System.Math
Imports Microsoft.SqlServer.Dts.Runtime

Public Class ScriptMain

Public Sub Main()

    Dim sVersion As String = Dts.Variables("sVersion").Value.ToString
    Dim sPackageLoadID As String = Dts.Variables("iPackageLoadID").Value.ToString
    Dim sMsg As String

sMsg = "Version: " & sversion & vbCrLf
sMsg = sMsg & "PackageLoadID: " & sPackageLoadID

MsgBox(sMsg)

Dts.TaskResult = Dts.Results.Success

End Sub

End Class

Execute the package to test the new script:

Next, add a new stored procedure named ssis.usp_RecordPackageEnd to the SSISRunTimeMetrics database using the following Transact-SQL script:

use SSISRunTimeMetrics
go

if exists(select s.name + '.' + p.name
          from sys.procedures p
          inner join sys.schemas s on s.schema_id = p.schema_id
          where p.name = 'usp_RecordPackageEnd'
          and s.name = 'ssis')
begin
Drop Procedure ssis.usp_RecordPackageEnd
end
go

Create Procedure ssis.usp_RecordPackageEnd
@id int
,@packageEndDateTime datetime = null
,@packageStatus varchar(25) = null
As
begin

-- set @packageEndDateTime default...
declare @Now datetime
set @Now = GetDate()
if (@packageEndDateTime Is Null)
set @packageEndDateTime = @Now

-- update the run time metrics data...
update ssis.RunTimeMetrics
Set packageEndDateTime = @packageEndDateTime
,packageStatus = @packageStatus
where id = @id

end

Now add a new Execute SQL Task named "Log End of Package Execution":

Set the Connection property to your SSISRunTimeMetrics database and the SQLStatement property to the following Transact-SQL statement:

declare @Now datetime
set @Now = GetDate()

Exec ssis.usp_RecordPackageEnd ?,@Now,'Succeeded'

Click the Parameter Mapping page and configure the iPackageLoadID variable as an input variable of the Long data type, assigned to ordinal 0:

Click the OK button:

Execute the package. Note that each time a change is saved, the VersionBuild property auto-increments, thus updating your sVersion variable value:

The package succeeds:

Examine the data stored in the ssis.RunTimeMetrics table. Note the latest row now contains a Start and End DateTime, and the packageStatus field now records "Succeeded":

We will work with these concepts more - fleshing them out by adding more robust code and additional functionality. For now, this is a good start. Congratulations!

Get the code!

:{> Andy