Data Transmission Service：使用ETL分析即時訂單

應用情境

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

• 多地區或異構資料即時集中：將多地區或者異構資料即時儲存至同一資料庫中，便於公司應用程式中心化高效管理及決策支援。

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

• Realtime Compute情境：對業務側產生的流資料即時清洗處理，形成特徵值、標籤支援線上業務計算模型（畫像、風控、推薦等）或即時大屏等Realtime Compute情境。

案例背景

本案例將為您示範如何使用流式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. 登入Data Transmission Service控制台。

   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中。