為 InfoSphere Warehouse 提供即時資料的高效整合方案：Replication Server + DataStage

前言

傳統 ETL 全量抽取在龐大資料量下效能低落（如 50GB 資料庫僅 0.01% 變更仍需全量掃描）。本文介紹結合 InfoSphere Replication Server（CDC 捕獲變更）與 InfoSphere DataStage（強大 ETL 轉換）的架構，實現資料倉庫即時更新，僅抽取變更資料，大幅提升效率。

核心組件介紹

InfoSphere Replication Server

企業級高速資料移動軟體，透過讀取資料庫恢復日誌捕獲提交的交易變更（Insert/Update/Delete）。兩種模式：

• SQL 複寫：Capture 寫入暫存表，Apply 套用至目標表。
• Q 複寫：高吞吐低延遲，捕獲變更轉為 WebSphere MQ 訊息傳遞，無暫存表。

Event Publisher (EP)：捕獲變更轉為 XML 或 CSV 訊息發布至 MQ 隊列，供下游應用（如 DataStage）直接消費，無需 Apply 程式。

InfoSphere DataStage

圖形化 ETL 工具，特點：

• 無代碼開發，支援複雜轉換邏輯重用。
• 廣泛資料源連接（DB2、Oracle、SQL Server、SAP、平面檔、XML、FTP、JMS 等）。
• 強大並行處理（分割、管道）提升效能。
• 支援批次與即時處理。

整合架構原理

1. Replication Server EP 的 Q Capture 從恢復日誌捕獲變更，寫入 MQ 隊列（CSV 格式）。
2. MQ 觸發器啟動 DataStage 作業。
3. DataStage 透過 MQ Connector Stage 讀取隊列訊息，解析 CSV，完成轉換後載入資料倉庫。

實作步驟詳解

環境需求

• IBM InfoSphere Replication Server 9.7
• IBM InfoSphere Information Server 8.1 (含 DataStage)

1. 來源端建立 DB2 物件與測試資料

CREATE TABLE "DEMO"."CUSTOMER" (
  "CUSTOMER_ID" INTEGER NOT NULL,
  "SEX" CHAR(1), "BIRTHDAY" TIMESTAMP, "SSN" VARCHAR(30),
  "CITY" VARCHAR(25), "STATE" VARCHAR(25), "ZIP" VARCHAR(15),
  "PHONE" VARCHAR(15), "PRI_LANGUAGE" VARCHAR(15),
  "LAST_UPDATE" TIMESTAMP,
  "FIRST_NAME" VARCHAR(20), "MIDDLE_NAME" VARCHAR(10), "LAST_NAME" VARCHAR(20)
);
ALTER TABLE "DEMO"."CUSTOMER" ADD PRIMARY KEY ("CUSTOMER_ID");
DB2 IMPORT FROM customer.ixf OF IXF INSERT INTO "DEMO"."CUSTOMER";

2. 建立 WebSphere MQ 物件

crtmqm QManager
strmqm QManager
runmqsc QManager < mq.in

mq.in 內容：

DEFINE QLOCAL(ADMINQ)
DEFINE QLOCAL(RESTARTQ)
DEFINE QLOCAL(q1)
DEFINE QMODEL(IBMQREP.SPILL.MODELQ) DEFSOPT(SHARED) MAXDEPTH(500000) MSGDLVSQ(FIFO) DEFTYPE(PERMDYN)
DEFINE CHANNEL(CHANNEL1) CHLTYPE(SVRCONN) TRPTYPE(TCP) MCAUSER('mqm')
DEFINE LISTENER(LISTENER1) TRPTYPE(TCP) CONTROL(QMGR) PORT(2264)
START LISTENER(LISTENER1)

3. 設定 Event Publisher

3.1 建立控制表：

asnclp -f cncap.in
# cncap.in 內容：
asnclp session set to q replication;
set run script now stop on sql error on;
set qmanager "QManager" for capture schema;
set server capture to db source;
create control tables for capture server using restartq "RESTARTQ" adminq "ADMINQ";
quit;

3.2 建立發佈佇列對應與發佈：

asnclp -f crtqmappub.in
# crtqmappub.in 內容：
asnclp session set to Q replication;
set output capture script "qpubmap.sql";
set log "qpub.log";
set server capture to db source;
set run script now stop on sql error on;
set qmanager "QManager" for capture schema;
create pubqmap pubqmap1 using sendq "q1" message format delimited message content type T;
create pub using pubqmap PUBQMAP1 (pubname eventpub1 "DEMO"."CUSTOMER");
quit;

3.3 啟動 Capture：

asnqcap capture_server=source capture_schema=ASN

日誌顯示發佈 EVENTPUB1 成功啟動。

4. 產生測試變更資料

CONNECT TO SOURCE;
INSERT INTO "DEMO"."CUSTOMER" VALUES (9000, 'F', '1960-05-06-00.00.00.000000', '467897085', 'Boise','Idaho','83701-83733','(71)657-9085','English', '2006-08-18-21.52.29.000000','IRVING','H','STERN');
INSERT INTO "DEMO"."CUSTOMER" VALUES (9100, 'F', '1960-05-06-00.00.00.000000', '467897085','Boise','Idaho','83701-83733','(71)657-9085','English', '2006-08-18-21.52.29.000000','IRVING','H','STERN');
UPDATE "DEMO"."CUSTOMER" SET CUSTOMER_ID=5000 WHERE CUSTOMER_ID=2075;
UPDATE "DEMO"."CUSTOMER" SET FIRST_NAME='meggy' WHERE FIRST_NAME='EILEEN';
UPDATE "DEMO"."CUSTOMER" SET LAST_NAME='Leee' WHERE LAST_NAME='JOHNSON';
DELETE FROM "DEMO"."CUSTOMER" WHERE CUSTOMER_ID=6057;
DELETE FROM "DEMO"."CUSTOMER" WHERE CUSTOMER_ID=8354;
CONNECT RESET;

5. MQ 訊息格式解析

EP 產生 CSV 訊息範例：

10,"IBM","2010098","181602875000","DEMO","CUSTOMER","ISRT","0000:0000:0000:0000:5914","0000:0000:0000:04f2:d60f","2010-04-08-17.16.02",,0000,,,,,,,,,,,,,,9000,"F","1960-05-06-00.00.00.000000","467897085","Boise","Idaho","83701-83733","(71)657-9085","English","2006-08-18-21.52.29.000000","IRVING","H","STERN"

10,"IBM","2010098","181603343000","DEMO","CUSTOMER","REPL","0000:0000:0000:0000:5916","0000:0000:0000:04f2:daaf","2010-04-08-17.16.02",,0000,2075,"F","1950-04-16-20.45.00.000000","205674075","Atlantaa","Georgia","30302-30399","(206)567-6075","English","2007-04-19-20.45.00.000000","SALLY","","STERN",5000,"F","1950-04-16-20.45.00.000000","205674075","Atlantaa","Georgia","30302-30399","(206)567-6075","English","2007-04-19-20.45.00.000000","SALLY","","STERN"

10,"IBM","2010098","181603343000","DEMO","CUSTOMER","DLET","0000:0000:0000:0000:591a","0000:0000:0000:04f2:f768","2010-04-08-17.16.02",,0000,8354,"F","2009-07-06-00.00.00.000000","34124","Beijing","fadsfs","10083","13439097809","Chinese","2009-07-28-00.00.00.000000","","","SHI LIXIAN",,,,,,,,,,,,,

關鍵欄位：第 5 欄 Schema、第 6 欄 Table、第 7 欄操作類型（ISRT/REPL/DLET）、後續欄位為變更前後資料。

6. DataStage 作業開發

6.1 MQ Connector Stage 設定

• 連線模式：Server（同機）或 Client（跨機）。
• 指定 Queue Manager、Queue 名稱、訊息格式 CSV。

6.2 Transformer Stage 邏輯處理

• 使用 Field() 函數擷取第 7 欄操作類型，Trim() 去除引號，存入變數 Operator。
• 定義三條輸出連結對應 ISRT、REPL、DLET，各自映射對應欄位資料。

6.3 後續處理

依操作類型分流後，分別連接至資料倉庫目標表的 Insert/Update/Delete 操作，完成即時同步。

結語

結合 Replication Server Event Publisher 的 CDC 能力與 DataStage 的強大轉換引擎，構建「唯讀日誌捕獲 → MQ 非同步傳遞 → 圖形化 ETL 處理」的即時資料管線。此架構避免全量掃描，僅處理變更資料，極大降低來源端負載與網路頻寬，為 InfoSphere Warehouse 提供高效、即時的資料餵養方案。