DataWorks：DataWorks快速入門

步驟一：資料同步

1. 建立資料來源。

   DataWorks通過建立資料來源的方式，接入資料來源和資料去向，因此，本步驟需要分別建立MySQL和MaxCompute兩個資料來源。

   • MySQL資料來源，用於串連資料來源（儲存業務資料的MySQL資料庫），為本教程提供原始業務資料。

     說明

     您無需準備本教程使用的原始業務資料，為方便測試和學習，DataWorks為您提供測試資料集，相關表資料已儲存在公網MySQL資料庫中，您只需建立MySQL資料來源接入即可。

     建立MySQL資料來源步驟

     1. 進入管理中心頁面。

        登入DataWorks控制台，切換至目標地區後，單擊左側導覽列的管理中心，在下拉框中選擇對應工作空間後單擊進入管理中心。

     2. 在左側導覽列單擊資料來源 > 資料來源列表，然後單擊新增資料來源，選擇資料來源類型為MySQL資料來源。

     3. 配置MySQL資料來源相關參數。

        說明

        • 表中未說明的參數保持預設值即可。

        • 首次新增資料來源時，需要完成跨服務授權，請根據頁面提示，授權服務關聯角色AliyunDIDefaultRole。

        參數	描述
        資料來源名稱	本樣本為MySQL_Source。
        配置模式	選擇串連串模式。
        串連地址	單擊新增地址。 主機地址IP：rm-bp1z69dodhh85z9qa.mysql.rds.aliyuncs.com 連接埠號碼：3306。 重要 本教程提供的資料僅作為阿里雲巨量資料開發治理平台 DataWorks資料應用實操使用，所有資料均為測試資料，並且僅支援在Data Integration模組讀取資料。
        資料庫名稱	配置為retail_e_commerce。
        使用者名稱	輸入使用者名稱workshop。
        密碼	輸入密碼workshop#2017。

     4. 在串連配置地區，找到工作空間已綁定的資源群組，單擊連通狀態列的測試連通性。

        說明

        如果MySQL資料來源連通性測試失敗，請檢查是否為資源群組綁定的VPC配置EIP，MySQL資料來源需要資源群組具備公網訪問能力。

     5. 單擊完成建立。

   • MaxCompute資料來源，用於串連資料去向（MaxCompute數倉），將MaxCompute資料來源綁定至資料開發後，能夠為本教程提供資料存放區和計算能力。

     • 如果您的專案空間中存在已添加的MaxCompute資料來源，則無需進行此步驟。

     • 如果您的專案空間中沒有添加MaxCompute資料來源，在參考如下步驟建立。

       建立MaxCompute資料來源步驟

       1. 在管理中心左側導覽列單擊資料來源 > 資料來源列表，然後單擊新增資料來源，選擇資料來源類型為MaxCompute資料來源。

       2. 配置MaxCompute資料來源相關參數。

          說明

          表中未說明的參數保持預設值即可。

          參數	描述
          資料來源名稱	本樣本為MaxCompute_Source。
          資料來源建立方式	選擇建立MaxCompute專案。
          專案名稱	自訂。 說明 MaxCompute專案名稱全域唯一，如後續測試連通性不通過，可能是因為專案名稱已存在，請修改專案名稱後再試。
          計算資源付費類型	選擇隨用隨付。 說明 如果隨用隨付不可選，請單擊隨用隨付後的去開通，完成MaxCompute服務的開通。
          預設Quota	下拉選擇預設已存在的Quota。
          預設訪問身份	阿里雲主帳號。

       3. 在串連配置地區，找到工作空間已綁定的資源群組，單擊連通狀態列的測試連通性。

          說明

          MaxCompute專案名稱全域唯一，如測試連通性不通過，報錯提示為ErrorMessage:[ODPS-0420095: Access Denied - Authorization Failed [4002], You don't exist in project xxx.，是因為專案名稱已存在，請修改專案名稱後再試。

       4. 單擊完成建立。

2. 資料開發綁定MaxCompute資料來源。

   需要先將MaxCompute資料來源綁定至資料開發，後續才能在資料開發模組中對MaxCompute的資料進行處理。

   1. 在左上方單擊 > 全部產品 > 資料開發與營運 > DataStudio（資料開發）。

   2. 在左側導覽列單擊資料來源（），找到已建立的MaxCompute資料來源，單擊綁定。

      說明

      如果您的資料開發模組已綁定了MaxCompute資料來源，則無需進行此步驟。

      

3. 建立虛擬節點，用於統籌管理整個電商平台銷售資料分析的商務程序。該節點為空白跑任務，無須編輯代碼。

   在左側導覽列單擊資料開發，找到商務程序 > Workflow，然後右鍵Workflow，選擇建立節點 > 通用 > 虛擬節點，自訂節點名稱，本教程以Workshop為例。

4. 建立離線同步任務。

   本教程使用的測試資料涉及兩張表（商品資訊源表item_info和訂單資訊源表trade_order），這兩張表格儲存體於MySQL資料來源關聯的MySQL資料庫中，本步驟需要分別建立兩個離線同步節點（節點名稱以ods_item_info和ods_trade_order為例），用於將item_info表和trade_order表同步至MaxCompute資料來源關聯的MaxCompute數倉中，然後再進行後續資料開發。

   1. 建立ods_item_info離線同步節點

      1. 右鍵Workflow，選擇建立節點 > Data Integration > 離線同步。

      2. 在建立節點頁面，設定名稱為ods_item_info。

      3. 配置資料來源為已建立的MySQL資料來源，資料去向為已綁定至資料開發的MaxCompute資料來源，資源群組選擇工作空間已綁定的資源群組，然後測試連通性都為可連通後，單擊下一步。

         

      4. 配置資料來源和資料去向的相關參數。

         說明

         表中未說明的參數保持預設值即可。

         配置地區	參數	ods_item_info離線同步節點取值
         資料來源	表	item_info
         資料去向	表	單擊一鍵產生目標表結構，在建立表視窗中，修改建表語句，將表名修改為ods_item_info，然後單擊建立表。
         通道控制	髒資料策略	配置為不容忍髒資料。

      5. 單擊離線同步任務配置視窗頂部的儲存（），然後單擊帶參運行（）。

         重要

         在離線計算情境下，bizdate為業務交易發生的日期，也常被稱為業務日期（business date）。例如，今天統計前一天的營業額，此處的前一天，指交易發生的日期，也就是業務日期。帶參運行時，自訂參數取值保持預設（目前時間的前一天）即可。

         如果您在此時修改了bizdate取值，請在後續本文所有節點涉及帶參運行時，同步修改bizdate取值。

   2. 建立ods_trade_order離線同步節點

      1. 右鍵Workflow，選擇建立節點 > Data Integration > 離線同步。

      2. 在建立節點頁面，設定名稱為ods_trade_order。

      3. 配置資料來源為已建立的MySQL資料來源，資料去向為已綁定至資料開發的MaxCompute資料來源，資源群組選擇工作空間已綁定的資源群組，然後測試連通性都為可連通後，單擊下一步。

      4. 配置資料來源和資料去向的相關參數。

         說明

         表中未說明的參數保持預設值即可。

         配置地區	參數	ods_trade_order離線同步節點取值
         資料來源	表	trade_order
         資料去向	表	單擊一鍵產生目標表結構，在建立表視窗中，修改建表語句，將表名修改為ods_trade_order，然後單擊建立表。
         通道控制	髒資料策略	配置為不容忍髒資料。

      5. 單擊離線同步任務配置視窗頂部的儲存（），然後單擊帶參運行（）。

步驟二：資料清洗

資料已從MySQL同步至MaxCompute後，獲得兩張資料表（商品資訊表ods_item_info和訂單資訊表ods_trade_order），您可以在DataWorks的資料開發模組對錶中資料進行清洗、處理和分析，從而擷取每日最暢銷商品類目排名表。

1. 建立dim_item_info節點。

   基於ods_item_info表，處理商品維度資料，產出商品基礎資訊維度資料表dim_item_info。

   操作步驟

   1. 右鍵Workflow，選擇建立節點 > MaxCompute > ODPS SQL，在建立節點頁面，設定名稱為dim_item_info後，單擊確認，配置如下SQL。

      CREATE TABLE IF NOT EXISTS dim_item_info (
          gmt_modified                   STRING COMMENT '商品最後修改日期',
          gmt_create                     STRING COMMENT '商品建立時間',
          item_id                        BIGINT COMMENT '商品數字ID',
          title                          STRING COMMENT '商品標題',
          sub_title                      STRING COMMENT '商品子標題',
          pict_url                       STRING COMMENT '主圖URL',
          desc_path                      STRING COMMENT '商品描述的路徑',
          item_status                    BIGINT COMMENT '商品狀態1：確認通過0：未確認通過',
          last_online_time               DATETIME COMMENT '最近一次開始銷售時間，商品上架時間',
          last_offline_time              DATETIME COMMENT '銷售結束時間，表示一個銷售周期的結束，僅作用於拍賣商品',
          duration                       BIGINT COMMENT '有效期間，銷售周期，只有兩個值，7天或14天',
          reserve_price                  DOUBLE COMMENT '當前價格',
          secure_trade_ordinary_post_fee DOUBLE COMMENT '平郵費用',
          secure_trade_fast_post_fee     DOUBLE COMMENT '快遞費用',
          secure_trade_ems_post_fee      DOUBLE COMMENT 'EMS郵費',
          last_online_quantity           BIGINT COMMENT '商品最近一次上架時的庫存數量',
          features                       STRING COMMENT '商品特徵',
          cate_id                        BIGINT COMMENT '商品葉子類目ID',
          cate_name                      STRING COMMENT '商品葉子類目名稱',
          commodity_id                   BIGINT COMMENT '品類ID',
          commodity_name                 STRING COMMENT '品類名稱',
          is_virtual                     STRING COMMENT '是否虛擬商品',
          shop_id                        BIGINT COMMENT '商家ID',
          shop_nick                      STRING COMMENT '商家NICK',
          is_deleted                     BIGINT COMMENT '類目是否刪除'
      )
      COMMENT '商品基礎資訊維度資料表'
      PARTITIONED BY (pt STRING COMMENT '業務日期, yyyymmdd')
      LIFECYCLE 365;
      
      
      -- 插入資料到 dim_item_info 表
      INSERT OVERWRITE TABLE dim_item_info PARTITION(pt='${bizdate}')
      SELECT
          gmt_create,
          gmt_modified,
          item_id,
          title,
          sub_title,
          pict_url,
          desc_path,
          item_status,
          last_online_time,
          last_offline_time,
          duration,
          cast(reserve_price as DOUBLE),
          cast(secure_trade_ordinary_post_fee as DOUBLE),
          cast(secure_trade_fast_post_fee as DOUBLE),
          cast(secure_trade_ems_post_fee as DOUBLE),
          last_online_quantity,
          features,
          cate_id,
          cate_name,
          commodity_id,
          commodity_name,
          is_virtual,
          shop_id,
          shop_nick,
          is_deleted
      FROM ods_item_info
      WHERE pt = '${bizdate}';

   2. 單擊ODPS SQL配置視窗頂部的儲存（），然後單擊帶參運行（）。

2. 建立dwd_trade_order節點。

   基於ods_trade_order表，對訂單的詳細交易資料進行初步清洗、轉換和商務邏輯處理，產出交易下單明細事實表dwd_trade_order。

   操作步驟

   1. 右鍵Workflow，選擇建立節點 > MaxCompute > ODPS SQL，在建立節點頁面，設定名稱為dwd_trade_order後，單擊確認，配置如下SQL。

      CREATE TABLE IF NOT EXISTS dwd_trade_order (
          id               BIGINT COMMENT '主鍵，去重後的最新id',
          gmt_create       DATETIME COMMENT '建立時間',
          gmt_modified     DATETIME COMMENT '修改時間',
          sub_order_id     BIGINT COMMENT '子訂單ID',
          parent_order_id  BIGINT COMMENT '父訂單ID',
          buyer_id         BIGINT COMMENT '買家數字id',
          buyer_nick       STRING COMMENT '買家暱稱，處理空值',
          item_id          BIGINT COMMENT '商品數字id',
          item_price       DECIMAL(38,18) COMMENT '商品價格，單位分',
          buy_amount       BIGINT COMMENT '購買數量',
          biz_type         BIGINT COMMENT '交易類型',
          memo             STRING COMMENT '備忘，處理空值',
          pay_status       BIGINT COMMENT '支付狀態',
          logistics_status BIGINT COMMENT '物流狀態',
          status           BIGINT COMMENT '狀態',
          seller_memo      STRING COMMENT '賣家的給交易的備忘',
          buyer_memo       STRING COMMENT '買家給交易的備忘',
          clean_ip         STRING COMMENT '清洗後的買家IP，過濾無效格式',
          end_time         DATETIME COMMENT '交易結束時間',
          pay_time         DATETIME COMMENT '付款的時間',
          is_sub           BIGINT COMMENT '是否是子訂單1表示子訂單',
          is_parent        BIGINT COMMENT '是否是父訂單1表示父訂單',
          shop_id          BIGINT COMMENT '商家id',
          total_fee        DECIMAL(38,18) COMMENT '去除折扣和調整後的子訂單費用',
          is_large_order_flag BOOLEAN COMMENT '是否為大額訂單標誌'
      )
      COMMENT '交易下單明細事實表，包含初步清洗和商務邏輯處理'
      PARTITIONED BY (pt STRING COMMENT '業務日期, yyyymmdd')
      LIFECYCLE 365; -- 資料生命週期設定為365天
      
      
      INSERT OVERWRITE TABLE dwd_trade_order PARTITION(pt='${bizdate}')
      SELECT
          MAX(id) AS id, -- 假設使用最新的id作為去重標準
          gmt_create,
          gmt_modified,
          sub_order_id,
          parent_order_id,
          buyer_id,
          COALESCE(buyer_nick, '') AS buyer_nick, -- 處理buyer_nick為空白的情況
          item_id,
          item_price,
          buy_amount,
          biz_type,
          COALESCE(memo, '') AS memo, -- 處理memo為空白的情況
          pay_status,
          logistics_status,
          status,
          seller_memo,
          buyer_memo,
          CASE 
              WHEN ip LIKE '__.__.__.__' THEN NULL -- 過濾無效IP格式
              ELSE ip 
          END AS clean_ip,
          end_time,
          pay_time,
          is_sub,
          is_parent,
          shop_id,
          total_fee,
          CASE 
              WHEN total_fee >= 10000 THEN TRUE -- 假設大於10000分的訂單為大額訂單
              ELSE FALSE 
          END AS is_large_order_flag -- 添加商務邏輯標誌
      FROM (
          SELECT
              *,
              ROW_NUMBER() OVER(PARTITION BY buyer_id, item_id, gmt_create ORDER BY id DESC) AS rn -- 用於去重的行號
          FROM ods_trade_order
          WHERE pt = '${bizdate}'
      ) AS sub_query
      WHERE rn = 1 -- 僅保留每個去重組的第一條記錄
      GROUP BY 
          gmt_create,
          gmt_modified,
          sub_order_id,
          parent_order_id,
          buyer_id,
          buyer_nick,
          item_id,
          item_price,
          buy_amount,
          biz_type,
          memo,
          pay_status,
          logistics_status,
          status,
          seller_memo,
          buyer_memo,
          clean_ip,
          end_time,
          pay_time,
          is_sub,
          is_parent,
          shop_id,
          total_fee,
          is_large_order_flag;

   2. 單擊ODPS SQL配置視窗頂部的儲存（），然後單擊帶參運行（）。

3. 建立dws_daily_category_sales節點。

   基於dwd_trade_order表和dim_item_info表，對DWD層經過清洗和標準化的詳細資料進行匯總，產出每日商品類目銷售匯總表dws_daily_category_sales。

   操作步驟

   1. 右鍵Workflow，選擇建立節點 > MaxCompute > ODPS SQL，在建立節點頁面，設定名稱為dws_daily_category_sales後，單擊確認，配置如下SQL。

      CREATE TABLE IF NOT EXISTS dws_daily_category_sales (
          cate_id             BIGINT COMMENT '商品葉子類目ID',
          cate_name           STRING COMMENT '商品葉子類目名稱',
          total_sales_amount  DECIMAL(38,18) COMMENT '商品類目總銷售額，單位分',
          order_count         BIGINT COMMENT '訂單數量'
      )
      COMMENT '每日商品類目銷售匯總表'
      PARTITIONED BY (pt STRING COMMENT '業務日期, yyyymmdd')
      LIFECYCLE 365;
      
      
      INSERT OVERWRITE TABLE dws_daily_category_sales PARTITION(pt='${bizdate}')
      SELECT
          i.cate_id,
          i.cate_name,
          SUM(t.total_fee) AS total_sales_amount,
          COUNT(DISTINCT t.id) AS order_count
      FROM dwd_trade_order t
      JOIN dim_item_info i ON t.item_id = i.item_id AND t.pt = i.pt
      WHERE t.pt = '${bizdate}'
      GROUP BY t.pt, i.cate_id, i.cate_name;

   2. 單擊ODPS SQL配置視窗頂部的儲存（），然後單擊帶參運行（）。

4. 建立ads_top_selling_categories節點。

   基於dws_daily_category_sales表，產出每日最暢銷商品類目排名表ads_top_selling_categories。

   操作步驟

   1. 右鍵Workflow，選擇建立節點 > MaxCompute > ODPS SQL，在建立節點頁面，設定名稱為ads_top_selling_categories後，單擊確認，配置如下SQL。

      CREATE TABLE IF NOT EXISTS ads_top_selling_categories (
          rank                BIGINT COMMENT '銷量排名',
          cate_id             BIGINT COMMENT '商品葉子類目ID',
          cate_name           STRING COMMENT '商品葉子類目名稱',
          total_sales_amount  DECIMAL(38,18) COMMENT '商品類目總銷售額，單位分',
          order_count         BIGINT COMMENT '訂單數量'
      )
      COMMENT '每日最暢銷商品類目排名表'
      PARTITIONED BY (pt STRING COMMENT '業務日期, yyyymmdd');
      
      
      INSERT OVERWRITE TABLE ads_top_selling_categories PARTITION(pt='${bizdate}')
      SELECT
          rank,
          cate_id,
          cate_name,
          total_sales_amount,
          order_count
      FROM (
          SELECT
              DENSE_RANK() OVER(ORDER BY total_sales_amount DESC) AS rank,
              cate_id,
              cate_name,
              total_sales_amount,
              order_count
          FROM (
              SELECT
                  cate_id,
                  cate_name,
                  SUM(total_sales_amount) AS total_sales_amount,
                  SUM(order_count) AS order_count
              FROM dws_daily_category_sales
              WHERE pt = '${bizdate}'
              GROUP BY cate_id, cate_name
          ) agg_sub
      ) agg_outer
      WHERE rank <= 10;

   2. 單擊ODPS SQL配置視窗頂部的儲存（），然後單擊帶參運行（）。

步驟三：資料展示

您已經將從MySQL中擷取的原始測試資料，經過資料開發處理，匯總於表ads_top_selling_categories中，現在可查詢表資料，查看資料分析後的結果。

1. 在左上方單擊 > 全部產品 > 資料分析 > SQL查詢。

2. 在我的檔案後單擊 > 建立檔案，自訂檔案名稱後單擊確定。

   

3. 在SQL查詢頁面，配置如下SQL。

   SELECT * FROM ads_top_selling_categories WHERE pt=${bizdate};

4. 單擊頂部的運行（），根據頁面提示，在右上方選擇MaxCompute資料來源後單擊確定，然後在費用預估頁面，單擊運行。

5. 在查詢結果中單擊，查看可視化圖表結果，您可以單擊圖表右上方的自訂圖表樣式。

   

6. 您也可以單擊圖表右上方儲存，將圖表儲存為卡片，然後在左側導覽列單擊卡片（）查看。

   

步驟四：周期性調度

通過完成前文操作步驟，您已經擷取了前一天各類商品的銷售資料，但是，如果需要每天擷取最新的銷售資料，則可以為資料開發中各任務節點配置周期任務，使其周期性定時執行。

1. 在左上方單擊 > 全部產品 > 資料開發與營運 > DataStudio（資料開發）。

2. 雙擊商務程序Workflow，在畫布中移動各節點位置並按下圖拖拽出各節點的上下遊依賴關係。

   

3. 單擊右側流程參數，配置參數名稱為bizdate，參數值或運算式為$bizdate，單擊儲存。

   

4. 雙擊虛擬節點（Workshop），配置如下周期調度參數後，單擊頂部的儲存（）。

   說明

   其他參數保持預設即可。

   

5. 切換至Workflow商務程序頁簽，單擊頂部的運行，參數bizdate填寫為前一天（例如今天為20240731，則此處填寫為20240730），測試所有流程是否均能成功運行。

   

6. 所有節點均能成功運行後，點擊頂部的提交，將流程中所有節點提交至營運中心。

   

7. 在左上方單擊 > 全部產品 > 資料開發與營運 > 營運中心（工作流程）。

8. 在周期任務營運 > 周期任務中即可看到已建立的周期任務。

   說明

   如需展示如下圖的所有上下遊依賴節點，請按右鍵Workshop節點，選擇展開子節點 > 四層。