開篇語

在這篇日志中?如何在 ETL 項(xiàng)目中統(tǒng)一管理上百個(gè) SSIS 包的日志和包配置框架?我介紹到了包級(jí)別的日志管理框架，那么這個(gè)主要是針對(duì)包這一個(gè)層級(jí)的 Log 信息，包括包開始執(zhí)行和結(jié)束時(shí)間，以及各個(gè)包的執(zhí)行成功或者失敗狀態(tài)。

但是我們可以更加深一層次的將日志記錄 Logging 以及數(shù)據(jù)信息 Auditing 信息延伸到包中的重要 Task 中。

通常情況下，SSIS 包從各個(gè)數(shù)據(jù)源加載數(shù)據(jù)到 Staging 表中，數(shù)據(jù)源可以是文件，也可以是其它數(shù)據(jù)庫。然后經(jīng)過數(shù)據(jù)倉庫 SCD 以及 Lookup 等操作，將 Staging 中的數(shù)據(jù)清理并整理加載到各個(gè)維度以及事實(shí)表中。

假設(shè)我們需要知道在當(dāng)前操作中，各個(gè) Staging 表加載了多少數(shù)據(jù)，使用了多長時(shí)間。各個(gè)處理維度和事實(shí)的 Task 使用了多少時(shí)間，新增了多少數(shù)據(jù)，修改了多少數(shù)據(jù)。這些我們也是有能力做到的，如果再配合?如何在 ETL 項(xiàng)目中統(tǒng)一管理上百個(gè) SSIS 包的日志和包配置框架?這篇文章中提到的包級(jí)別日志記錄，那么我們將非常清晰的知道我們的 SSIS 包無論是在包級(jí)別，還是在各個(gè)重要 Task 級(jí)別的各種日志，數(shù)據(jù)信息。這些信息對(duì)于我們的包維護(hù)，性能分析，錯(cuò)誤糾正，錯(cuò)誤修復(fù)都是非常有價(jià)值的。

比如，我可以很輕松的通過自定義的報(bào)表瀏覽哪些 Task 在同等記錄情況下最耗時(shí)間，各個(gè) Task 在整個(gè)包的執(zhí)行過程中所用的時(shí)間比。

不同的 ETL 項(xiàng)目在 Auditing 上會(huì)采取不同的策略，比如以文件加載為主的 ETL 是允許有部分錯(cuò)誤數(shù)據(jù)加載失敗的，但是以數(shù)據(jù)倉庫為主的 ETL 則不希望出現(xiàn)錯(cuò)誤數(shù)據(jù)加載的。因此在設(shè)計(jì) Auditing 的時(shí)候要考慮到這些情況，比如設(shè)計(jì)的時(shí)候多出一個(gè) 失敗數(shù)據(jù)總數(shù)的記錄用于跟蹤文件數(shù)據(jù)等。

在這里只講如何實(shí)現(xiàn) Auditing，簡(jiǎn)單的介紹一下核心操作，大家可以在這個(gè)基礎(chǔ)之上去擴(kuò)充。

關(guān)鍵點(diǎn)

實(shí)現(xiàn) Auditing 的關(guān)鍵點(diǎn)就是要借用控制流 Task 中的 Event Handler 下的 OnPostExecute 和 OnPreExecute 功能。

OnError 功能已經(jīng)在這篇?如何在 ETL 項(xiàng)目中統(tǒng)一管理上百個(gè) SSIS 包的日志和包配置框架?文章中詳細(xì)的介紹到了。

OnPreExecute - 在 Task 執(zhí)行之前觸發(fā)的事件。

OnPostExecute - 在 Task 執(zhí)行完成之后觸發(fā)的事件。

通過這樣兩個(gè)事件我們很容易實(shí)現(xiàn)對(duì) Task 執(zhí)行前后表數(shù)據(jù)變化的操作記錄。

數(shù)據(jù)源，目標(biāo)表及其它數(shù)據(jù)庫對(duì)象

測(cè)試數(shù)據(jù)源是 AdventureWorksLT2012

BIWORK_SSIS 數(shù)據(jù)庫中的目標(biāo)表?

IF OBJECT_ID('dbo.SalesOrderDetail') IS NOT NULL DROP TABLE dbo.SalesOrderDetail GOCREATE TABLE [dbo].[SalesOrderDetail]([SalesOrderID] [int] NOT NULL,[SalesOrderDetailID] [int] NOT NULL,[OrderQty] [smallint] NOT NULL,[ProductID] [int] NOT NULL,[UnitPrice] [money] NOT NULL,[UnitPriceDiscount] [money] NOT NULL,[LineTotal] [numeric](38, 6) NOT NULL,[rowguid] [uniqueidentifier] NOT NULL,[ModifiedDate] [datetime] NOT NULL ) ON [PRIMARY] GO

Task 執(zhí)行狀態(tài)表

EXECUTION_ID 應(yīng)該使用?如何在 ETL 項(xiàng)目中統(tǒng)一管理上百個(gè) SSIS 包的日志和包配置框架?中的 PROCESS_LOG_ID, 這樣就將 SSIS 包日志和 TASK 關(guān)聯(lián)起來了。

USE BIWORK_SSIS GOIF OBJECT_ID('TASK_EXECUTION_STATUS') IS NOT NULL DROP TABLE TASK_EXECUTION_STATUS GOCREATE TABLE TASK_EXECUTION_STATUS (EXECUTION_ID NVARCHAR(255), PACKAGE_NAME NVARCHAR(100),TASK_ID NVARCHAR(250),TASK_NAME NVARCHAR(250),TABLE_NAME NVARCHAR(250),ExistingRowsBefore BIGINT,StartTime DATETIME,DeletedRows BIGINT,UpdatedRows BIGINT,InsertedRows BIGINT,ExistingRowsAfter BIGINT,EndTime DATETIME,ExecutionStatus INT )

獲取表的條數(shù)

IF OBJECT_ID('dbo.GET_TABLE_COUNT') IS NOT NULL DROP PROCEDURE dbo.GET_TABLE_COUNT GOCREATE PROCEDURE dbo.GET_TABLE_COUNT @TABLE_NAME NVARCHAR(50), @ROW_COUNT BIGINT OUTPUT AS BEGIN SELECT @ROW_COUNT = SUM(PART.rows) FROM sys.tables TBLINNER JOIN sys.partitions PART ON TBL.object_id = PART.object_idINNER JOIN sys.indexes IDX ON PART.object_id = IDX.object_idAND PART.index_id = IDX.index_idWHERE TBL.name = @TABLE_NAMEAND IDX.index_id < 2GROUP BY TBL.object_id, TBL.name RETURN @ROW_COUNT END GO

記錄時(shí)間

這個(gè)存儲(chǔ)過程用來每次在執(zhí)行 Task 之前獲取目標(biāo)表中的條數(shù)，并且插入 Task 啟動(dòng)時(shí)間 -

IF OBJECT_ID('dbo.USP_INSERT_TASK_EXECUTION','P') IS NOT NULL DROP PROCEDURE dbo.USP_INSERT_TASK_EXECUTION GOCREATE PROCEDURE USP_INSERT_TASK_EXECUTION@TARGET_TABLE_NAME NVARCHAR(50),@EXECUTION_ID NVARCHAR(255) , @PACKAGE_NAME NVARCHAR(100),@TASK_ID NVARCHAR(255),@TASK_NAME NVARCHAR(250) AS BEGINDECLARE @ExistingRowsBefore BIGINTEXECUTE dbo.GET_TABLE_COUNT@TABLE_NAME = @TARGET_TABLE_NAME,@ROW_COUNT = @ExistingRowsBefore OUTPUT INSERT INTO TASK_EXECUTION_STATUS(EXECUTION_ID, PACKAGE_NAME,TASK_ID,TASK_NAME,TABLE_NAME,ExistingRowsBefore,StartTime,DeletedRows,UpdatedRows,InsertedRows,ExistingRowsAfter,EndTime,ExecutionStatus)VALUES(@EXECUTION_ID, @PACKAGE_NAME,@TASK_ID,@TASK_NAME,@TARGET_TABLE_NAME,@ExistingRowsBefore,GETDATE(),NULL, --@DeletedRows,NULL, --@UpdatedRows,NULL, --@InsertedRows,NULL, --@ExistingRowsAfterNULL, --@EndTime0 -- In process ) END

更新 Task 狀態(tài)表?

當(dāng) @DeletedRows = -1 的時(shí)候，表明操作是 Truncate 操作。

IF OBJECT_ID('dbo.USP_UPDATE_TASK_EXECUTION' ) IS NOT NULL DROP PROCEDURE dbo.USP_UPDATE_TASK_EXECUTION GOCREATE PROCEDURE dbo.USP_UPDATE_TASK_EXECUTION @ExecutionID NVARCHAR(250), @TaskID NVARCHAR(250), @DeletedRows BIGINT, @UpdatedRows BIGINT, @InsertedRows BIGINT AS BEGINUPDATE dbo.TASK_EXECUTION_STATUSSET DeletedRows = (CASE WHEN @DeletedRows = -1 THEN ExistingRowsBefore ELSE @DeletedRows END),UpdatedRows = (CASE WHEN @DeletedRows = -1 THEN 0 ELSE @UpdatedRows END),InsertedRows = (CASE WHEN @DeletedRows = -1 THEN 0 ELSE @InsertedRows END),ExistingRowsAfter = (CASE WHEN @DeletedRows = -1 THEN 0 ELSE (ExistingRowsBefore + @InsertedRows - @DeletedRows) END),EndTime = GETDATE(),ExecutionStatus = 1WHERE EXECUTION_ID = @ExecutionIDAND TASK_ID = @TaskID END GO

SSIS 包中的流程實(shí)現(xiàn)

SSIS 包 - 第二個(gè)和第三個(gè) Task 的功能完全一樣，只為了演示的目的。

EST_TRUNCATE_SALES_ORDER_DETAIL Task - 在加載數(shù)據(jù)之前刪除表數(shù)據(jù)。

它的 OnPreExecute 事件中添加了一個(gè) Execute SQL Task 組件用來向 Task Execution 表插入操作前的記錄。

調(diào)用 USP_INSERT_TASK_EXECUTION 存儲(chǔ)過程根據(jù)表名查詢記錄數(shù)。

參數(shù) Mapping 關(guān)系，注意這里要用 Source ID , Source Name 而不是 Task ID, Task Name。因?yàn)?Task 是指當(dāng)前執(zhí)行這些 SQL 的 Task 自身，而我們要監(jiān)控的不是這個(gè)事件下的 Task ，而是控制流中的 Task 組件。

EST_TRUNCATE_SALES_ORDER_DETAIL 中 OnPostExecute 的配置 -

這里就是 Update 操作了，因?yàn)?EST_TRUNCATE_SALES_ORDER_DETAIL?是 Truncate 表操作，所以這里給了 DeletedRows = -1。

更新的時(shí)候直接根據(jù) Execution ID 和 Task ID 就可以了。

第二個(gè) Task 也要配置 OnPreExecute 和 OnPostExecute 事件，也就是說每一個(gè)你需要監(jiān)控的 Task 都要配置。感覺比較復(fù)雜，但是一次配置完成以后，受用可是長期的。

要注意的是第二個(gè) Task 是從數(shù)據(jù)源加載數(shù)據(jù)，這樣需要在加載的過程中獲取記錄數(shù)，通過 ROW COUNT 可以實(shí)現(xiàn)將數(shù)據(jù)流的條數(shù)賦值給變量保存。

另外要注意的是 - 這個(gè)變量的 SCOPE 是控制流組件自身，即作用域。因?yàn)榭赡芤泻芏?Task 需要用到記錄條數(shù)的變量，全部放到包級(jí)別中這個(gè)變量會(huì)非常多，并且容易出錯(cuò)。可以理解為 InsertedRows 是局部變量，它的生命周期就是 Task 本身。

如果創(chuàng)建的變量位于包級(jí)別 SCOPE，可以點(diǎn)擊下方的小方框 Move?到當(dāng)前 Task 的 SCOPE 中。

變量的賦值。

OnPreExecute 的配置和上面的 Task 一樣，復(fù)制一份即可，這里是 OnPostExecute 的配置。

需要什么變量就記錄什么變量，就配置什么變量。

后面的兩個(gè) Task 一模一樣，只是為了測(cè)試使用。

運(yùn)行兩次的結(jié)果，數(shù)據(jù)條數(shù)的記錄是非常連貫的。

記錄 SCD 的修改和新增條數(shù)只需要在相應(yīng)的地方添加 ROW COUNT 組件來捕獲即可。

當(dāng)然除了使用 ROW COUNT 組件，在某些特定的情況下也可以使用 @@ROWCOUNT 來獲取新增，刪除或者修改所影響到的條數(shù)。

DECLARE @UpdateRowCnt INT DECLARE @InsertRowCnt INT--Inserting records from Source to Destination which does not exists insert into dbo.Client(ClientName,Country,Town) Select clientName, Country, Town from dbo.ClientSource S WHERE NOT EXISTS ( Select 1 from dbo.Client CL WHERE CL.ClientName=S.ClientName) SELECT @InsertRowCnt=@@ROWCOUNT--Update Already existing records from Source Update CL set CL.ClientType=S.CLientType from dbo.Client CL INNER JOIN dbo.ClientSource S ON Cl.ClientName=S.ClientName SELECT @UpdateRowCnt=@@ROWCOUNT

最后一個(gè)問題

如果每次都在各個(gè) Task 中的 OnPreExecute 和 OnPostExecute 中配置非常麻煩，有沒有改進(jìn)的方法。

答案是有的。

我提供一個(gè)思路，有興趣的話可以自動(dòng)動(dòng)手嘗試 -

Task 級(jí)別的 OnPreExecute 和 OnPostExecute 事件是當(dāng) Task 被執(zhí)行前后被觸發(fā)的，要注意的是包級(jí)別的 OnPreExecute 和 OnPostExecute 也是可以捕獲 Task 級(jí)別的 OnPreExecute 和 OnPostExecute 事件。

可以定義一張表，表中記錄需要被處理的 Task 名稱，然后在包級(jí)別的 OnPreExecute 和 OnPostExecute 中處理 各個(gè) Task 的 Auditing 信息。不在列表上的，就可以不用處理。

同時(shí)還要注意 Task 同步的問題，若是很多 Task 同時(shí)執(zhí)行，并行執(zhí)行的話，就需要在各自 Task 中定義好變量來記錄然后再賦值給 Package 級(jí)別的變量可以避免這一問題。

與本文相關(guān)的文章

如何在 ETL 項(xiàng)目中統(tǒng)一管理上百個(gè) SSIS 包的日志和包配置框架

更多 BI 文章請(qǐng)參看?BI 系列隨筆列表 (SSIS, SSRS, SSAS, MDX, SQL Server)????? 如果覺得這篇文章看了對(duì)您有幫助，請(qǐng)幫助推薦，以方便他人在 BIWORK 博客推薦欄中快速看到這些文章。

總結(jié)

以上是生活随笔為你收集整理的如何管理和记录 SSIS 各个 Task 的开始执行时间和结束时间以及 Task 中添加|删除|修改的记录数...的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。