應用場景

為滿足企業處理實時數據的需求，ETL提供了流式數據抽取、加工和加載功能，能夠高效整合海量實時數據，支持拖拽式操作和低代碼開發方式，幫助企業輕松完成商業決策分析、報表提速、實時數據計算等。企業在數字化轉型過程中，涉及實時數據處理的應用場景如下：

• 多區域或異構數據實時集中：將多地域或者異構數據實時存儲至同一數據庫中，便于企業中心化高效管理及決策支持。

• 報表提速：幫助客戶構建實時報表體系，不僅大幅提升報表產出效率，還能支持更多實時分析場景，滿足了企業數字化轉型階段對報表產出效率的高要求。

• 實時計算場景：對業務側產生的流數據實時清洗處理，形成特征值、標簽支持在線業務計算模型（畫像、風控、推薦等）或實時大屏等流計算場景。

案例背景

本案例將為您演示如何使用流式ETL功能，將實時交易數據（訂單號、客戶ID、產品/商品編碼、交易金額、交易時間）與業務維度數據（產品編碼、產品單價、產品名稱等）相結合，并將滿足過濾條件的數據（如統計單筆超3000的實時交易信息）實時集中至數據倉庫，實現交易數據的多維分析（如產品維度、客戶維度等）。您還可根據業務需要，借助工具實現可視化大屏，洞察動態數據。

任務配置流程

準備工作

在配置ETL任務前，您需將實時交易數據和業務維度數據分別作為流表和維表存儲在源RDS MySQL數據庫中。

并根據業務需求，在目標RDS MySQL數據庫中建表。

create table test_orders(
   order_id bigint not null COMMENT '訂單ID',
     user_id bigint not null comment '用戶ID',
     product_id bigint not null comment '產品ID',
   total_price decimal(15,2) not null COMMENT '訂單總額',
   order_date TIMESTAMP  not null COMMENT '訂單日期',
   PRIMARY KEY (order_id))

CREATE table product (
      product_id bigint not null comment '產品ID',
      product_name varchar(20) comment '產品名稱',
      product_price decimal(15,2) not null comment '產品單價')

create table test_orders(
 order_id bigint not null COMMENT '訂單ID',
 user_id bigint not null comment '用戶ID',
 product_id bigint not null comment '產品ID',
 total_price decimal(15,2) not null COMMENT '訂單總額',
 order_date TIMESTAMP  not null
 COMMENT '訂單日期',
 product_id_2 bigint not null comment '產品ID',
 product_name varchar(20) comment '產品名稱',
 product_price decimal(15,2) not null comment '產品單價',
 PRIMARY KEY (order_id))

步驟一：配置源庫信息

1. 進入ETL任務的列表頁面。

   1. 登錄數據傳輸服務DTS控制臺。

   2. 在左側導航欄，單擊ETL。

2. 單擊左上角的，在新增數據流對話框中，您需在數據流名稱配置ETL任務名稱，選擇開發方式為DAG。

3. 單擊確認。

4. 執行如下操作，配置流表和維表信息。

   1. 配置流表信息

      1. 頁面左側，將輸入/維表 MySQL節點拖拽至頁面右側畫布的空白區域。

      2. 單擊畫布區域的輸入/維表 MySQL-1，根據頁面信息配置源庫信息。

      3. 設置以下參數，配置節點信息。

         參數	說明
         請輸入數據源名稱	DTS會自動生成一個數據源名稱，建議配置具有業務意義的名稱（無唯一性要求），便于后續識別。
         區域	選擇源庫所在地域。 說明 當前僅支持在華東1（杭州）、華東2（上海）、華北1（青島）、華北2（北京）、華北3（張家口）、華南1（深圳）、華南3（廣州）和中國香港創建ETL任務。
         實例列表	選擇源庫所在數據庫實例的實例名稱。您也可以單擊下方的新建實例進行新建，新建方法請參見DMS支持的數據庫。
         節點類型	選擇源表的類型，本案例中選擇為流表。 流表：實時發生變化的表，可以關聯一個維表，用于數據關聯查詢。 維表：更新不頻繁（非實時更新）的表，一般用于結合實時數據拼裝成寬表進行數據分析。
         轉換格式	ETL在處理數據時會將流轉換為動態表，在該動態表上進行持續查詢（即動態表會被INSERT、UPDATE、DELETE操作持續更改），產生一個新的動態表。最終寫入目標庫時，再將新的動態表會轉化為流。當新的動態表轉化為流時，您需要指定轉化格式，對動態表前后更改信息進行編碼： Upsert流：動態表中的數據支持通過INSERT、UPDATE和DELETE操作修改，當轉換為流時，會將INSERT和UPDATE操作編碼為upsert message，將DELETE操作編碼為delete message。 說明 該編碼方式要求動態表具有唯一鍵（可能是復合的）。 Append-Only流： 動態表中的數據僅支持INSERT操作修改，當轉換為流時僅需發送INSERT的數據。
         庫表選擇	選擇源表中需轉換的庫表。

      4. 節點配置完成后，頁面會自動切換至輸出字段頁簽，您可根據業務需要，在頁簽的列名稱列勾選需要的字段。

      5. 本案例中選擇為流表，需要單擊時間屬性頁簽，并設置對應參數。

         參數	說明
         選擇事件時間Watermark字段	選擇流表中的一個時間字段。一般流表會定義時間字段，代表數據產生的時間，通常為具有業務含義的時間戳（比如ordertime）。
         事件時間Watermark延遲時間	輸入數據延遲的最大容忍時間。 應用場景是，由于數據并不一定按照實際產生順序，達到ETL等待處理，可能會出現延遲情況。如果數據一直延遲未到，ETL不能無限制地等待延遲的數據，因此需要建立延遲時間來處理亂序數據。比如10:00的數據已到達，但是9:59的數據還未到達，則ETL只會等待至“10:00+延遲時間”。如果9:59的數據在“10:00+延遲時間”后到達，則ETL會拋棄該數據。
         處理時間ProcTime	處理時間為ETL處理數據時的本地時間。您需要自定義一個列名，ETL會在該列保存數據處理的本地時間。處理時間主要用于算子運算，如時態JOIN會用該處理時間去關聯普通表的最新版本。

      說明

      完成配置的源庫右側不顯示時，說明配置完成。

   2. 配置維表信息

      1. 頁面左側，將輸入/維表 MySQL節點拖拽至頁面右側畫布的空白區域。

      2. 單擊畫布區域的輸入/維表 MySQL-2，根據頁面信息配置源庫信息。

      3. 設置以下參數，配置節點信息。

         參數	說明
         請輸入數據源名稱	DTS會自動生成一個數據源名稱，建議配置具有業務意義的名稱（無唯一性要求），便于后續識別。
         區域	選擇源庫所在地域。
         實例列表	選擇源庫所在數據庫實例的實例名稱。您也可以單擊下方的新建實例進行新建，新建方法請參見DMS支持的數據庫。
         節點類型	選擇源表的類型，本案例中選擇為維表。
         庫表選擇	選擇源表中需轉換的庫表。

      4. 節點配置完成后，頁面會自動切換至輸出字段頁簽，您可根據業務需要，在頁簽的列名稱列勾選需要的字段。

      說明

      完成配置的源庫右側不顯示時，說明配置完成。

步驟二：配置表JOIN

1. 在頁面左側，將表 Join節點拖拽至頁面右側畫布的空白區域。

2. 將鼠標指針移動至已完成配置的流表和維表上，單擊圓點拉出連接線，分別將流表和維表與表 Join-1連接起來。

3. 單擊畫布區域的表 Join-1，根據頁面信息配置轉換組件。

   1. 在節點配置頁簽，設置以下參數，配置節點信息。

      

      區域	參數	說明
      轉換名稱	請輸入轉換名稱	DTS會自動生成一個轉換組件名稱，建議配置具有業務意義的名稱（無唯一性要求），便于后續識別。
      Join配置	Join類型符左邊的表	選擇放置在JOIN類型符左邊的表，作為主表。本案例中選擇為流表。
      	時態Join時間屬性（不選擇為普通Join）	選擇使用時態JOIN時，流表關聯時態表的時間屬性。如不輸入，則默認使用普通JOIN。本案例中選擇為基于處理時間ProcTime。 說明 時態表，也稱動態表，是指基于表的（參數化）視圖概念，根據時間記錄數據變更歷史，分為版本表（可顯示數據的歷史版本）和普通表（僅顯示數據的最新版本）。 時態JOIN要求流表定義時間屬性，右表要有主鍵；如右表是維表，則您所設置的Join的條件需包含維表的主鍵。 基于事件時間Watermark：使用流表的事件時間去關聯版本表對應的版本。 基于處理時間ProcTime：使用流表的處理時間去關聯普通表的最新版本。
      	選擇Join操作	選擇Join操作方式。本案例中選擇為Inner Join。 Inner Join：數據為兩張表的交集。 Left Join：在左表中獲取所有數據，在右表中獲取兩張表的交集。 Right Join：在左表中獲取兩張表的交集，在右表中獲取所有數據。
      Join條件	+新增條件	單擊+新增條件，選擇JOIN的條件字段。 說明 等號（=）左側為JOIN后新表的左表字段，右側為JOIN后新表的右表字段。

4. 完成Join條件配置后，單擊輸出字段頁簽。根據實際需要，在列名稱列勾選需要的字段。

步驟三：配置表記錄過濾

1. 在頁面左側，將表記錄過濾節點拖拽至頁面右側畫布的空白區域。

2. 將鼠標指針移動至已完成配置的表 Join-1上，單擊圓點拉出連接線，連接表 Join-1和表記錄過濾-1。

3. 單擊畫布區域的表記錄過濾-1，根據頁面信息配置轉換組件。

   1. 在轉換名稱區域，輸入轉換名稱。

      說明

      DTS會自動生成一個轉換組件名稱，建議配置具有業務意義的名稱（無唯一性要求），便于后續識別。

   2. 在where條件區域，您可以選擇以下任意一種方法配置WHERE條件。

      • 直接輸入需要的WHERE條件，比如輸入total_price>3000.00，表示過濾JOIN后表中total_price字段的值大于3000.00的數據。

      • 單擊字段輸入或操作符區域中的選項配置WHERE條件。

步驟四：配置目標庫信息

1. 在頁面左側，將輸出 MySQL節點拖拽至頁面右側畫布的空白區域。

2. 將鼠標指針移動至已完成配置的表記錄過濾-1上，單擊圓點拉出連接線，連接表記錄過濾-1和輸出 MySQL-1。

3. 單擊畫布區域的輸出 MySQL-1頁面，根據頁面信息配置目標庫信息。

   1. 設置以下參數，配置節點信息。

      

      參數	說明
      請輸入數據源名稱	DTS會自動生成一個數據源名稱，建議配置具有業務意義的名稱（無唯一性要求），便于后續識別。
      區域	選擇目標庫所在地域。 說明 當前僅支持在華東1（杭州）、華東2（上海）、華北1（青島）、華北2（北京）、華北3（張家口）、華南1（深圳）、華南3（廣州）和中國香港創建ETL任務。
      實例列表	選擇目標庫所在數據庫的實例名稱。您也可以單擊下方的新建實例進行新建，新建方法請參見DMS支持的數據庫。
      表映射	選擇目標庫通過轉換處理后需要存儲的目標表。 在選擇目標表區域，單擊目標表。

4. 根據業務需要，在列名稱列勾選需要的參數。

步驟五：預檢查并啟動任務

1. 配置完成后，單擊生成Flink SQL校驗，ETL將生成Flink SQL并進行校驗。

2. 檢驗完成后，您可單擊查看ETL校驗詳情，在彈跳框中，查看Flink SQL生成結果和SQL語句。確認無誤后，單擊關閉。

   說明

   如校驗失敗，您可以根據生成結果顯示的失敗原因進行修復。

3. 單擊下一步保存任務并預檢查。當預檢查通過后，DTS才能開始ETL任務。如果預檢查失敗，請單擊檢查失敗項后的查看詳情，根據提示信息修復后，重新進行預檢查。

4. 預檢查完成后，單擊頁面下方的下一步購買。

5. 在購買頁面，選擇鏈路規格和計算資源，閱讀并勾選數據傳輸（按量付費）服務條款和公測協議條款。

   說明

   公測期間，每個用戶可以免費創建并使用兩個ETL實例。

6. 單擊購買并啟動，ETL任務正式開始。

任務運行結果

本案例中，ETL任務的啟動后（以8月1日為例），如實時交易數據表test_orders中更新的數據滿足過濾條件（total_priceid>3000.00，即總交易額大于3000.00），則該數據會同步至目標表test_orders_new中。