Phần 3 - Tích hợp dữ liệu vào kho dữ liệu bằng công cụ SSIS

Discussion in 'Giải pháp ERP, CRM, EPM and BI' started by tableau, May 21, 2014.

  1. tableau

    tableau Well-Known Member

    3.1.Convert dữ liệu bằng SSIS
    3.1.1.Pack_Customer_Dim

    Ở trong Tab Control Flow kéo Data Flow Task ở Control Flow Items bên trái vào
    [​IMG]
    Ở trong Data Flow tab kéo OLE DB Source từ Data Flow Sources và OLE DB Destination từ Data Flow Destinations sau đó đổi tên thành bảng nguồn và bảng đích
    [​IMG]
    Tiếp theo tạo connection tới data source và data destination ở trong page Connection Managers[​IMG]
    Kéo link từ source tới destination và Edit Source, Trong Connection Manager Chọn OLE DB onnection manager chọn tới data source và chọn table như hình dưới:
    [​IMG]
    Trong Columns chọn những field cần lấy dữ liệu như hình dưới:
    [​IMG]
    Edit Destination, chọn connection tới data warehouse và chọn bảng đích như hình dưới:
    [​IMG]
    Và Mapping các cột data source với data warehouse như hình dưới:[​IMG]
    Kết quả sau khi chạy:
    [​IMG]
    Các bảng Dimension khác làm tương tự
    3.2.Chạy Real-time với SSIS
    Để dữ liệu chạy Real-time được trước hết phải bật chức năng CDC của tất cả các table cần bắt dữ liệu của SQL lên. Để Enable CDC được các table trước hết cần Enable CDC Database bằng câu lệnh:
    USE MicrosoftDynamicsAX
    GO
    EXEC sys.sp_cdc_enable_db
    GO
    Enable CDC Table bằng câu lệnh:
    USE MicrosoftDynamicsAX
    GO
    EXEC sys.sp_cdc_enable_table
    @source_schema =N'dbo',
    @source_name=N'CUSTTABLE',
    @role_name=null,
    @supports_net_changes = 1
    GO
    3.1.1.Pack_Customer_Dim
    Đầu tiên thiết kế các Control Flow như hình dưới:
    [​IMG]
     

    Attached Files:

    • 1.png
      1.png
      File size:
      27.4 KB
      Views:
      1,460
    • 2.png
      2.png
      File size:
      34.4 KB
      Views:
      1,419
    • 3.png
      3.png
      File size:
      3.7 KB
      Views:
      1,399
    • 4.png
      4.png
      File size:
      24.2 KB
      Views:
      1,413
    • 5.png
      5.png
      File size:
      29.2 KB
      Views:
      1,408
    • 6.png
      6.png
      File size:
      23.8 KB
      Views:
      1,368
    • 7.png
      7.png
      File size:
      23.2 KB
      Views:
      1,369
    • 8.png
      8.png
      File size:
      20.5 KB
      Views:
      1,389
    • 9.png
      9.png
      File size:
      21.7 KB
      Views:
      1,385
    • 10.png
      10.png
      File size:
      20.4 KB
      Views:
      1,352
  2. Loading...


  3. tableau

    tableau Well-Known Member

    • Caculation Interval
    Sử dụng Execute SQL Task tính toán thời gian bắt đầu và kết thúc để lấy dữ liệu:
    Mở SSIS, trong cửa sổ Variables tạo 2 biến ExtractStartTime và ExtractEndTime kiểu Datetime
    Mở Exucute SQL Task Editor, trên General page, ở ResultSet chọn Single row, ở connection cấu hình kết nối tới data source, ở SQL Source Type chọn Direct input, ở SQL Statement viết câu lệnh lấy thời gian bắt đầu và kết thúc:
    SELECT DATEADD(dd,0, DATEDIFF(dd,0,GETDATE()-2))AS ExtractStartTime,

    DATEADD(dd,0, DATEDIFF(dd,0,GETDATE()-1)) AS ExtractEndTime
    [​IMG]
    Trên Result Set mapping biến ExtractStartTime và ExtractEndTime, như hình dưới:


    [​IMG]
    • Wait for capture process
    Vòng lặp này sẽ thực hiện lặp để chờ xem trong khoảng thời gian StartTime và EndTime có dữ liệu sẵn sàng để lấy hay không, và sẽ lặp chờ bao lâu đó nếu không có dữ liệu sẽ báo lỗi hoặc báo timeout
    Tạo biến DataReady kiểu int, giá trị ban đầu đặt = 0 để lưu trạng thái trả về của Execute SQL Task.Tạo biến DelaySeconds kiểu int, giá trị ban đầu = 10 để lưu khoảng thời gian cho phép trễ
    Tạo biến TimeoutCount kiểu int, giá trị ban đầu = 0 để lưu giá trị của vòng lặp hiện tại
    Tạo biến TimeoutCeiling kiểu int, giá trị ban đầu = 20 để lưu số lần vòng lặp sẽ thực hiện trước khi trả về kết quả lỗi hoặc hết thời gian
    Tạo biến IntervalID kiểu int có giá trị = 0 để lưu là giá trị đầu tiên của vòng lặp
    Trên For Loop Editor, trên InitExpression và EvalExpression ghi vào @DataReady == 0
    [​IMG]
    Trong vòng lặp thêm một Execute SQL Task. Task này queries CDC process và trả về tình trạng của dữ liệu được lưu trong DataReady.
    Giá trị mà DataReady trả về:
    0: CDC không sẵn sàng -> tiếp tục thực hiện Delay và lặp lại vòng lặp
    1: CDC không được bắt trong khoảng thời gian này hoặc đã được xóa -> tiếp tục thực hiện Log Error
    2: CDC sẵn sàng -> pass qua vòng lặp và thực hiện Task tiếp theo
    3: CDC sẵn sàng từ lần tải đầu tiên -> pass qua vòng lặp và thực hiện Task tiếp theo
    5: Đạt đến TimeoutCeiling -> ngừng lại
    Trong Execute SQL Task Editor, trên General page, Ở Result Set chọn Single row, ở connection cấu hình kết nối tới nguồn dữ liệu, ở SQL Source Type chọn Direct input, ở SQL Statement:
    declare @DataReady int, @TimeoutCount int
    if not exists(select tran_end_time from cdc.lsn_time_mapping
    where tran_end_time > ? )
    select @DataReady = 0
    else
    if ? = 0
    select @DataReady = 3
    else
    ifnotexists(select tran_end_time from cdc.lsn_time_mapping
    where tran_end_time <= ? )
    select @DataReady = 1
    else
    select @DataReady = 2
    select @TimeoutCount = ?
    if (@DataReady = 0)
    select @TimeoutCount = @TimeoutCount + 1
    else
    select @TimeoutCount = 0
    if (@TimeoutCount > ?)
    select @DataReady = 5
    select @DataReady as DataReady, @TimeoutCount as TimeoutCount[​IMG]
    Mapping các biến như hình bên dưới:
    [​IMG]
    Mapping output trong Result Set:
    [​IMG]
    • DELAY
    Trong vòng lặp thêm Script Task, mở Precedence Constraint Editor, trong Evaluation operation, chọn Expression and Constraint, trong Value, chọn Success, trong Expression, ghi vào @DataReady == 0 && @TimeoutCount <= @TimeoutCeiling Chọn Logical AND. All constraints must evaluate to True nếu nó chưa được chọn. Trong Script Task Editor, Trên Script page, ở ReadOnlyVariables, chọn biến User::DelaySeconds, Click Edit Script ghi câu lệnh vào phần main code
    System.Threading.Thread.Sleep((int)Dts.Variables["DelaySeconds"].Value * 1000);
    • ERROR
    Trong vòng lặp thêm Script Task, mở Precedence Constraint Editor, trong Evaluation operation, chọn Expression and Constraint, trong Value, chọn Success, trong Expression, ghi vào @DataReady == 1 || @DataReady == 5 Chọn Logical AND. All constraints must evaluate to True nếu nó chưa được chọn. Trong Script Task Editor, Trên Script page, ở ReadOnlyVariables, chọn biến User::DataReady User::ExtractStartTime, Click Edit Script ghi câu lệnh vào phần main code để show error
    [​IMG]
     

    Attached Files:

    • 11.png
      11.png
      File size:
      11.1 KB
      Views:
      1,371
    • 12.png
      12.png
      File size:
      9.4 KB
      Views:
      1,341
    • 13.png
      13.png
      File size:
      17.3 KB
      Views:
      1,341
    • 14.png
      14.png
      File size:
      14 KB
      Views:
      1,363
    • 15.png
      15.png
      File size:
      9.9 KB
      Views:
      1,318
    • 16.png
      16.png
      File size:
      19.5 KB
      Views:
      1,332
    • 17.png
      17.png
      File size:
      35.7 KB
      Views:
      1,324
    • 18.png
      18.png
      File size:
      18.7 KB
      Views:
      1,337
    • 19.png
      19.png
      File size:
      11.7 KB
      Views:
      1,328
    • 20.png
      20.png
      File size:
      9.5 KB
      Views:
      1,330
    tableau, tableau and tableau like this.
  4. tableau

    tableau Well-Known Member

    • Prepare Query
    Trước khi chuẩn bị query CDC ta phải tạo Table-valued Funtion bằng câu lệnh:
    create table #wrapper_functions
    (function_name sysname, create_stmt nvarchar(max))
    insert into #wrapper_functions
    exec sys.sp_cdc_generate_wrapper_function
    declare @stmt nvarchar(max)
    declare #hfunctions cursor local fast_forward for
    select create_stmt from #wrapper_functions
    open #hfunctions
    fetch #hfunctions into @stmt
    while (@@fetch_status <> -1)
    begin
    execsp_executesql@stmt
    fetch #hfunctions into @stmt
    end
    close #hfunctions
    deallocate #hfunctions
    [​IMG]

    Tạo một biến SqlDataQuery kiểu String để chứa chuỗi giá trị từ câu query, giá trị mặc định là
    select * from [fn_all_changes_dbo_CUSTTABLE](null,null,'all')
    TrongExecute SQL Task Editor, trênGeneralpage, Ở ResultSet, chọnSingle row, Ở SQLSourceType chọnDirect input, SQL Statement:
    declare @ExtractStartTime datetime,@ExtractEndTime datetime,
    @DataReady int
    select @DataReady = ?,
    @ExtractStartTime = ?,
    @ExtractEndTime = ?
    if @DataReady = 2
    select N'select * from [fn_all_changes_dbo_BSD$CUSTTABLE]'+ N'('''+ convert(nvarchar(30),@ExtractStartTime,120)+ ', '+ convert(nvarchar(30),@ExtractEndTime,120) + ',''all'')'as SqlDataQuery
    else
    select N'select * from [fn_all_changes_dbo_BSD$CUSTTABLE]'+ N'(null, '''+ convert(nvarchar(30),@ExtractEndTime,120)+ ''',''all'') ' as SqlDataQuery

    [​IMG]

    Mapping các biến như hình dưới:

    [​IMG]

    Mapping Result như hình dưới:

    [​IMG]
     
    tableau, tableau and tableau like this.
  5. tableau

    tableau Well-Known Member

    • Extract Data
    Trong Data Flow thiết kế các Control Flow như hình dưới:
    [​IMG]
    Trong OLE DB Source Editor Chọn Connection đến data source, Data access mode là SQL command from variable, Ở Variable name chọn User::SqlDataQuery.
    [​IMG]

    Column: chọn những cột cần lấy dữ liệu, check chọn như hình bên dưới
    [​IMG]
    Cấu hình Conditional Split để phân lo dữ liệu Insert, Update, Delete
    Trong Conditional Split Transformation Editor điền vào 3 dòng điều kiện như hình dưới:
    CDC_OPERATION == "I"
    CDC_OPERATION == "UN"
    CDC_OPERATION == "D"
    [​IMG]

    Edit Insert: Trong OLE DB Source Editor, ở Connection Manager page cấu hình giống như hình dưới:
    [​IMG]

    General NewID()

    Kéo một Script Component vào, chọn Transfomation
    [​IMG]

    Trong Script Transformation Editor chọn Ouput 0, Add column Name: NewID, Data Type unique identifier [DT_GUID] như hình dưới:
    [​IMG]
    Trong Script chọn Edit Script và ghi vào đoạn code
    Row.NewID = System.Guid.NewGuid();
    Mapping:
    [​IMG]
     

    Attached Files:

    • 21.png
      21.png
      File size:
      15.7 KB
      Views:
      1,322
    • 22.png
      22.png
      File size:
      16.5 KB
      Views:
      1,315
    • 23.png
      23.png
      File size:
      22.6 KB
      Views:
      1,314
    • 24.png
      24.png
      File size:
      18.8 KB
      Views:
      1,319
    • 25.png
      25.png
      File size:
      17.5 KB
      Views:
      1,382
    • 26.png
      26.png
      File size:
      8.8 KB
      Views:
      1,318
    • 27.png
      27.png
      File size:
      28.3 KB
      Views:
      1,288
    • 28.png
      28.png
      File size:
      23.2 KB
      Views:
      1,351
    tableau, tableau and tableau like this.
  6. tableau

    tableau Well-Known Member

    Edit Update: Trong Advanced Editor for OLE DB Command trên Connection Manager tab, chọn connection tới AX_DW
    [​IMG]
    Trên Component Properties tab,SqlCommand viết câu lệnh Update:
    [​IMG]
    Column Mapping: Các Parameter sẽ tương ứng với thứ tự các cột trong câu lệnh Update
    [​IMG]
    Edit Delete : Trong Advanced Editor for OLE DB Command trên Connection Manager tab, chọn connection tới AX_DW
    [​IMG]
    Trên Component Properties tab,SqlCommand viết câu lệnh Delete:
    [​IMG]
    Column Mapping:
    [​IMG]
     

    Attached Files:

    • 29.png
      29.png
      File size:
      10.9 KB
      Views:
      1,288
    • 30.png
      30.png
      File size:
      23 KB
      Views:
      1,273
    • 31.png
      31.png
      File size:
      19.9 KB
      Views:
      1,298
    • 32.png
      32.png
      File size:
      10.8 KB
      Views:
      1,280
    • 33.png
      33.png
      File size:
      23 KB
      Views:
      1,289
    • 34.png
      34.png
      File size:
      18.6 KB
      Views:
      1,240
    tableau and tableau like this.
  7. tableau

    tableau Well-Known Member

    • Clear_cdc
    Tạo biến count kiểu int, gán giá trị là 500
    Tạo biến cdc_table_name kiểu nvarchar(500), gán giá trị là table cần delete
    Tạo Store Clear_cdc_data bằng câu lệnh
    CREATE PROCEDURE [dbo].[cdc_clear_lsn](@cdc_table_name nvarchar(500), @count int, @min_lsn binary(10) output, @max_lsn binary(10) output)
    AS
    BEGIN
    SET NOCOUNT ON;
    declare @temp table (
    [__$start_lsn] binary(10)
    )
    declare @sql nvarchar(max)
    set @sql = 'select top '+cast(@count as nvarchar(10))+' [__$start_lsn] from [cdc].['+@cdc_table_name+'] order by [__$seqval]'
    insert into @temp
    exec (@sql)
    select @min_lsn = MIN([__$start_lsn]), @max_lsn = MAX([__$start_lsn]) from @temp
    declare @query nvarchar(2000)
    set @query = 'delete from [cdc].['+@cdc_table_name+'] where [__$start_lsn] <= @max_sq'
    Exec sp_executesql@query, N'@max_sq binary(10)', @max_sq = @max_lsn
    END
    Trong Execute SQL Task Editor, ở Result Set chọn None, ở SQL Statement ghi vào EXEC clear_cdc_data ?,?
    [​IMG]
    Trong Parameter Mapping:
    [​IMG]
    Kết quả sau khi chạy:
    [​IMG]
     

    Attached Files:

    • 35.png
      35.png
      File size:
      20.1 KB
      Views:
      1,282
    • 36.png
      36.png
      File size:
      10.8 KB
      Views:
      1,278
    • 37.png
      37.png
      File size:
      21.6 KB
      Views:
      1,262

Share This Page