步驟一：數據同步

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數據應用實操使用，所有數據均為測試數據，并且僅支持在數據集成模塊讀取數據。
        數據庫名稱	配置為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，選擇新建節點 > 數據集成 > 離線同步。

      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，選擇新建節點 > 數據集成 > 離線同步。

      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節點，選擇展開子節點 > 四層。