DataWorks：最佳實務：資料開發工作流程中配置資料推送節點

步驟一：搭建推送流程

雙擊開啟資料推送Demo流程商務程序圖頁面，依次將查詢上月銷售總額通過達標條件判斷分成達標上月銷售達標資料、推送銷售達標前三名類別和不達標上月銷售不達標資料、推送銷售不達標倒數三名類別，關聯起來，產生一條如下圖所示的條件推送的工作流程。

步驟二：配置SQL查詢節點

通過配置SQL查詢節點對測試資料進行查詢，通過節點上下文參數產生SQL查詢節點的輸出參數outputs，從而實現將SQL的查詢結果傳遞至分支節點。

1. 雙擊開啟查詢上月銷售總額節點，編寫以下查詢SQL代碼。

   -- 統計上個月銷售額
   SELECT SUM(sales) AS sales_amount
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59');

2. 編輯完成SQL代碼後，單擊右側調度配置，在彈出的調度配置面板中配置如下內容。

   • 定時調度時間：08：00。

   • 調度資源群組：選擇已建立好的Serverless資源群組。

   • 依賴上遊節點：勾選使用工作空間根節點。

   • 本節點輸出參數：在節點上下文參數下單擊本節點輸出參數後的添加賦值參數，添加輸出參數作為下遊節點輸入參數的取值。

步驟三：配置分支節點

在分支節點內，通過節點上下文參數的本節點輸入參數擷取上遊SQL查詢節點的輸出參數outputs，然後在分支節點進行條件判斷後通過本節點輸出參數輸出合格資料。

1. 雙擊開啟達標條件判斷節點，單擊添加分支，新增兩個分支，分別為達標和不達標，詳細配置資訊如下

   配置項	達標分支	不達標分支
   分支條件	${inputs[0][0]}>=500000	${inputs[0][0]}<500000
   關聯到節點輸出	達標。	不達標。
   分支描述	上月銷售總額達標。	上月銷售總額不達標。

   說明

   [0][0]為二維數組，用來定位擷取到的參數中用於條件判斷的資料。

   • 上遊為SQL查詢節點情況下，使用二維數組，定位參數中需要作為條件的資料。

   • 上遊為Python節點情況下，使用一維數組，定位參數中需要作為條件的資料。

2. 雙擊開啟達標條件判斷節點，單擊右側調度配置，在彈出的調度配置中配置如下內容。

   參數類型		參數內容與說明
   時間屬性	調度周期	日
   	定時調度時間	08：00
   	重跑屬性	運行成功或失敗後皆可重跑。
   資源屬性	調度資源群組	選擇已建立好的調度資源群組。 說明 初次使用資料推送節點，需先提工單升級調度資源群組。
   本節點輸出名稱		當配置分支後，輸出名會自動解析並添加到此處。輸出名與分支定義的關聯到節點輸出一致。
   節點上下文參數	本節點輸入參數	參數名：inputs 取值來源：選擇上遊節點查詢上月銷售總額的輸出參數outputs。
   	本節點輸出參數	系統預設添加為outputs。

3. 完成配置後，單擊儲存達標條件判斷節點。

步驟四：配置分支SQL查詢節點

分支SQL查詢節點在本實踐中分為上月銷售達標資料和上月銷售不達標資料兩個節點，需要將分支節點上下文參數輸出至上月銷售達標資料節點和上月銷售不達標資料節點，然後將分支節點SQL查詢結果用節點上下文參數的本節點輸出參數分別輸出至對應的達標與不達標的資料推送節點。

1. 分別雙擊上月銷售達標資料節點和上月銷售不達標資料節點，進入編輯頁面，編寫以下SQL代碼。

   上月銷售達標資料

   SET @all_cat_sales_volume_month := 0.0;
   SELECT SUM(sales) INTO @all_cat_sales_volume_month FROM orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59');
   
   -- 建立暫存資料表
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array (
     category VARCHAR(255),
     sales DOUBLE,
     all_cat_sales_volume_month DOUBLE
   );
   --查詢並寫入暫存資料表
   INSERT INTO temp_array (category, sales, all_cat_sales_volume_month) SELECT category, SUM(sales) AS amount, @all_cat_sales_volume_month FROM orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59') GROUP BY category ORDER BY amount DESC limit 3;
   --查詢達標前三名資料
   SELECT category, sales, all_cat_sales_volume_month FROM temp_array;

   上月銷售不達標資料

   SET @all_cat_sales_volume_month := 0.0;
   SELECT SUM(sales) INTO @all_cat_sales_volume_month FROM orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59');
   
   --建立暫存資料表
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array (
     category VARCHAR(255),
     sales DOUBLE,
     all_cat_sales_volume_month DOUBLE
   );
   
   --查詢並寫入暫存資料表
   INSERT INTO temp_array (category, sales, all_cat_sales_volume_month) SELECT category, SUM(sales) AS amount, @all_cat_sales_volume_month FROM orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59') GROUP BY category ORDER BY amount ASC limit 3;
   
   --查詢不達標的後三名資料
   SELECT category, sales, all_cat_sales_volume_month FROM temp_array;

2. 編輯完成SQL代碼後，單擊右側調度配置，在彈出的調度配置面板中配置如下內容。

   • 定時調度時間：08：00。

   • 調度資源群組：選擇已建立好的Serverless資源群組。

   • 依賴的上遊節點：添加上遊的分支節點，因在步驟一：搭建推送流程已建立好流程，需查看上遊依賴是否正確。

     • 上月銷售達標資料節點：上遊節點輸出名為達標。

     • 上月銷售不達標資料節點：上遊節點輸出名為不達標。

   • 本節點輸出參數：在節點上下文參數下單擊本節點輸出參數後的添加賦值參數，添加輸出參數作為下遊節點輸入參數的取值。

3. 單擊儲存上月銷售達標資料節點和上月銷售不達標資料節點。

步驟五：配置分支資料推送節點

需要建立兩個資料推送節點，分別配置節點上下文參數的本節點輸入參數，來擷取上月銷售達標資料和上月銷售不達標資料的分支SQL查詢節點輸出的outputs參數，在本文中使用這些參數並將其推送至各自的目標。

1. 雙擊推送銷售達標前三名類別節點和推送銷售不達標倒數三名類別節點，進入節點後，單擊調度配置，在調度配置面板中進行如下配置。

   參數類型		參數內容	圖示
   調度參數	參數名	curdate
   	參數值	$[yyyymmddhh:mi:ss]
   時間屬性	調度周期	日
   	定時調度時間	08：00 說明 本實踐以8:00為例，確保資料能在08:00時推送至目標渠道。您可以根據實際需要配置其他時間點。
   	重跑屬性	運行成功或失敗後皆可重跑。
   資源屬性	調度資源群組	選擇已建立好的調度資源群組。 說明 初次使用資料推送節點，需先提工單升級調度資源群組。
   節點上下文參數	本節點輸入參數	單擊添加新增本節點輸入參數： 參數名：inputs。 取值來源： 在推送銷售達標前三名類別節點選擇上月銷售達標資料節點的輸出參數outputs 在推送銷售不達標倒數三名類別節點選擇上月銷售不達標資料節點的輸出參數outputs	推送銷售達標前三名類別節點 推送銷售不達標倒數三名類別節點

2. 調度配置完成後，即可配置資料推送內容，詳情如下。

   • 資料推送目標：下拉資料推送目標選擇所需的資料推送目標，若不存在，可單擊下拉框右下角的建立資料推送目標，建立推送目標。

   參數	說明
   類型	支援圈選DingTalk、飛書、企業微信、Teams以及郵件。
   對象名稱	可按業務需求進行自訂。
   WebHook	DingTalk、飛書、企業微信機器人或Teams的Webhook，以及郵件的SMTP需要在相應的目標平台上擷取。 說明 飛書機器人Webhook擷取方式可參見：飛書Webhook觸發器配置。 Teams Webhook擷取方式可參見：使用 Microsoft Teams 工作流程建立傳入 Webhook。

   • 標題：推送銷售達標前三名類別和推送銷售不達標倒數三名類別。

   • 本文：按需求進行配置，詳情可參見配置推送內容。

     說明

     本文中可直接使用上遊SQL查詢節點查詢的欄位名作為預留位置，以擷取上遊輸入的參數。

     • 推送銷售達標前三名類別樣本

     • 推送銷售不達標倒數三名類別樣本

3. 配置完成後，單擊儲存推送銷售達標前三名類別節點和推送銷售不達標倒數三名類別節點。

步驟六：測試條件推送流程

在完成條件推送流程配置後，需要對條件測試流程進行測試，以便後續提交發布。

1. 雙擊開啟資料推送Demo流程商務程序頁面。

2. 選中查詢上月銷售總額節點，右鍵選擇運行節點及下遊，等待運行完成即可。

   說明

   如果出現任務運行失敗，選中需要查看的節點，單擊右鍵選擇查看日誌即可查看日誌。

   

步驟一：搭建推送流程

雙擊開啟資料推送Demo流程商務程序圖頁面，在商務程序圖頁面，可以看到已建立的多個節點，依次將查詢上一周銷售資料、組織銷售前三名列表及推送上周前三名類別進行關聯，產生一條如下圖所示的指令碼推送的工作流程。

步驟二：配置SQL查詢節點

通過配置SQL查詢節點對測試資料進行查詢，可以通過節點上下文參數產生SQL查詢節點的輸出參數outputs，從而實現將SQL的查詢結果傳遞至賦值節點。

1. 雙擊查詢上一周銷售資料節點，進入編輯頁面，編寫以下SQL代碼。

   -- 統計上一周銷售額
   SELECT category, SUM(sales) AS amount
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 WEEK), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 23:59:59')
   GROUP BY category ORDER BY amount DESC limit 3;

2. 編輯完成SQL代碼後，單擊右側調度配置，在彈出的調度配置面板中配置如下內容。

   • 定時調度時間：08：00。

   • 調度資源群組：選擇已建立好的Serverless資源群組。

   • 依賴上遊節點：勾選使用工作空間根節點。

   • 本節點輸出參數：在節點上下文參數下單擊本節點輸出參數後的添加賦值參數，添加輸出參數作為下遊節點輸入參數的取值。

3. 單擊儲存查詢上一周銷售資料節點。

步驟三：配置賦值節點

賦值節點通過節點上下文參數下的本節點輸入參數可以擷取到上遊SQL節點的輸出參數outputs，將SQL節點的輸出參數進行重新賦值調整，產生新的本節點輸出參數，將新資料輸出至資料推送節點。

1. 雙擊組織銷售前三名列表的賦值節點，進入節點編輯頁面，編寫以下Python代碼。

   def main():
   
       From datetime import date
       today = date.today()
       formatted_date = today.strftime('%Y-%m-%d')
       
       msg = 'Stat date: ' + formatted_date + ' \\n\\n ' \
       '- 1: ${inputs[0][0]}, sales: ${inputs[0][1]} \\n\\n ' \
       '- 2: ${inputs[1][0]}, sales: ${inputs[1][1]} \\n\\n ' \
       '- 3: ${inputs[2][0]}, sales: ${inputs[2][1]} \\n\\n '
       
       print(msg)
   
   
   if __name__ == "__main__":
       import sys
       main()

2. 編輯完成賦值代碼後，單擊右側調度配置，在彈出的調度配置面板中配置如下內容。

   • 定時調度時間：08：00。

   • 調度資源群組：選擇已建立好的調度資源群組。

   • 節點上下文參數。

     • 本節點輸入參數：

       • 參數名：inputs。

       • 取值來源：上遊查詢上一周銷售資料節點的輸出參數outputs。

     • 本節點輸出參數：由系統預設產生。

3. 單擊儲存組織銷售前三名列表賦值節點。

步驟四：配置資料推送節點

通過配置節點上下文參數的本節點輸入參數，來擷取上遊賦值節點輸出的outputs參數，在本文中使用這些參數並將其推送至目標。

1. 雙擊名為推送上周前三名類別的資料推送節點，進入節點後，單擊調度配置，在調度配置面板中進行如下配置。

   參數類型		參數內容	圖示
   調度參數	參數名	curdate。
   	參數值	$[yyyymmddhh:mi:ss]。
   時間屬性	調度周期	日。
   	定時調度時間	08：00。 說明 本實踐以8:00為例，確保資料能在08:00時推送至目標渠道。您可以根據實際需要配置其他時間點。
   	重跑屬性	運行成功或失敗後皆可重跑。
   資源屬性	調度資源群組	選擇已建立好的調度資源群組。 說明 初次使用資料推送節點，需先提工單升級調度資源群組。
   節點上下文參數	本節點輸入參數	單擊添加新增本節點輸入參數： 參數名：inputs。 取值來源：選擇上遊節點上周銷售前三類的輸出參數outputs。

2. 調度配置完成後，即可配置資料推送內容，詳情如下。

   • 資料推送目標：下拉資料推送目標選擇所需的資料推送目標，若不存在，可單擊下拉框右下角的建立資料推送目標，建立推送目標。

   參數	說明
   類型	支援圈選DingTalk、飛書、企業微信、Teams以及郵件。
   對象名稱	可按業務需求進行自訂。
   WebHook	DingTalk、飛書、企業微信機器人或Teams的Webhook，以及郵件的SMTP需要在相應的目標平台上擷取。 說明 飛書機器人Webhook擷取方式可參見：飛書Webhook觸發器配置。 Teams Webhook擷取方式可參見：使用 Microsoft Teams 工作流程建立傳入 Webhook。

   • 標題：推送上周前三名類別。

   • 本文：按需求進行配置本文即可，詳情可參見配置推送內容。

     說明

     本文中可直接使用上遊SQL查詢節點查詢的欄位名作為預留位置，以擷取上遊輸入的參數。

     

3. 配置完成後，單擊儲存推送上周前三名類別推送節點。

步驟五：測試指令碼推送流程

在完成指令碼推送流程配置後，需要對指令碼測試流程進行測試，以便後續提交發布。

1. 雙擊資料推送Demo流程開啟商務程序圖頁面。

2. 選中查詢上一周銷售資料節點，右鍵選擇運行節點及下遊，等待運行完成即可。

   說明

   如果出現任務運行失敗，選中需要查看的節點，單擊右鍵選擇查看日誌即可查看日誌。

   

步驟一：搭建推送流程

雙擊資料推送Demo流程後，在商務程序圖頁面，可以看到已建立的多個節點，需拖動MySQL查詢節點查詢昨日銷售總額作為上遊節點，並將其與下遊的資料推送節點推送昨日銷售總額關聯起來，產生一條如下圖所示的簡單推送的工作流程。

步驟二：配置SQL查詢節點

通過配置SQL查詢節點對測試資料進行查詢，可以通過節點上下文參數產生SQL查詢節點的輸出參數outputs，從而實現將SQL的查詢結果傳遞至資料推送節點。

1. 雙擊查詢昨日銷售總額節點，編寫以下查詢SQL代碼。

   -- 建立暫存資料表temp_array
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array (
     total_amount DOUBLE
   );
   
   -- 將查詢到的昨日銷售總額寫入暫存資料表temp_array中
   INSERT INTO temp_array (total_amount) 
   SELECT SUM(sales)
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 23:59:59');
   
   -- 查詢暫存資料表temp_array
   select total_amount FROM temp_array;

2. 編輯完成SQL代碼後，單擊右側調度配置，在彈出的調度配置面板中配置如下內容。

   • 定時調度時間：08：00。

   • 調度資源群組：選擇已建立好的Serverless資源群組。

   • 依賴上遊節點：勾選使用工作空間根節點。

   • 本節點輸出參數：在節點上下文參數下單擊本節點輸出參數後的添加賦值參數，添加輸出參數作為下遊節點輸入參數的取值。

3. 單擊儲存查詢昨日銷售總額節點。

步驟三：配置資料推送節點

通過配置節點上下文參數的本節點輸入參數，來擷取上遊SQL查詢節點輸出的outputs參數，在本文中使用這些參數並將其推送至目標。

1. 雙擊名為推送昨日銷售總額的資料推送節點，進入節點後，單擊調度配置，在調度配置面板中進行如下配置。

   參數類型		參數內容	圖示
   調度參數	參數名	curdate
   	參數值	$[yyyymmddhh:mi:ss]
   時間屬性	調度周期	日
   	定時調度時間	08:00 說明 本實踐以8:00為例，確保資料能在08:00時推送至目標渠道。您可以根據實際需要配置其他時間點。
   	重跑屬性	運行成功或失敗後皆可重跑。
   資源屬性	調度資源群組	選擇已建立好的調度資源群組。 說明 初次使用資料推送節點，需先提工單升級調度資源群組。
   節點上下文參數	本節點輸入參數	單擊添加新增本節點輸入參數： 參數名 : inputs。 取值來源：選擇上遊節點查詢昨日銷售總額的輸出參數outputs。

2. 調度配置完成後，即可配置資料推送內容，詳情如下。

   • 資料推送目標：下拉資料推送目標選擇所需的資料推送目標，若不存在，可單擊下拉框右下角的建立資料推送目標，建立推送目標。

     參數	說明
     類型	支援圈選DingTalk、飛書、企業微信、Teams以及郵件。
     對象名稱	可按業務需求進行自訂。
     WebHook	DingTalk、飛書、企業微信機器人或Teams的Webhook，以及郵件的SMTP需要在相應的目標平台上擷取。 說明 飛書機器人Webhook擷取方式可參見：飛書Webhook觸發器配置。 Teams Webhook擷取方式可參見：使用 Microsoft Teams 工作流程建立傳入 Webhook。

   • 標題：推送昨日銷售總額。

   • 本文：按需求進行配置本文即可，詳情可參見配置推送內容。

     說明

     本文中可直接使用上遊SQL查詢節點查詢的欄位名作為預留位置，以擷取上遊輸入的參數。

     

3. 配置完成後，單擊儲存推送昨日銷售總額推送節點。

步驟四：測試簡單推送流程

在完成簡單推送流程配置後，需要對簡單測試流程進行測試，以便後續提交發布。

1. 雙擊開啟資料推送Demo流程商務程序圖頁面。

2. 選中查詢昨日銷售總額節點，右鍵選擇運行節點及下遊，等待運行完成即可。

   說明

   如果出現任務運行失敗，選中需要查看的節點，單擊右鍵選擇查看日誌即可查看日誌。

   

步驟一：搭建推送流程

雙擊開啟資料推送Demo流程商務程序圖頁面，在商務程序圖頁面，可以看到已建立的多個節點，拖拽上遊節點查詢昨日銷售總額、查詢昨日銷售增長額與下遊節點推送昨日銷售總額與增長額關聯起來，產生一條如下圖所示的合并推送的工作流程。

步驟二：配置SQL查詢節點

通過配置SQL查詢節點對測試資料進行查詢，可以通過節點上下文參數產生SQL查詢節點的輸出參數outputs，將SQL的查詢結果傳遞至資料推送節點。

1. 單擊查詢昨日銷售增長額節點，編寫以下查詢SQL代碼。

   -- 建立統計前日資料表temp_array1
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array1 (
     category VARCHAR(255),
     sales DOUBLE
   );
   -- 寫入前日資料至temp_array1
   INSERT INTO temp_array1 (category, sales) SELECT category, SUM(sales)
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 2 DAY), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 2 DAY), '%Y-%m-%d 23:59:59')
   GROUP BY category;
   
   -- 建立統計昨日資料表temp_array2
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array2 (
     category VARCHAR(255),
     sales DOUBLE
   );
   -- 寫入昨日資料至temp_array2
   INSERT INTO temp_array2 (category, sales) SELECT category, SUM(sales)
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 23:59:59')
   GROUP BY category;
   
   -- 建立統計昨日銷售增長額result
   CREATE TEMPORARY TABLE IF NOT EXISTS result (
     category VARCHAR(255),
     diff DOUBLE
   );
   -- 寫入增長額資料至result
   INSERT INTO result (category, diff) SELECT temp_array2.category AS category, temp_array2.sales - temp_array1.sales AS diff FROM temp_array1 LEFT JOIN temp_array2 ON temp_array1.category = temp_array2.category;
   
   -- 查詢增長額暫存資料表result
   SELECT category, diff FROM result;

2. 編輯完成SQL代碼後，單擊右側調度配置，在彈出的調度配置面板中配置如下內容。

   • 定時調度時間：08：00。

   • 調度資源群組：選擇已建立好的Serverless資源群組。

   • 依賴上遊節點：勾選使用工作空間根節點。

   • 本節點輸出參數：在節點上下文參數下單擊本節點輸出參數後的添加賦值參數，添加輸出參數作為下遊節點輸入參數的取值。

3. 單擊儲存查詢昨日銷售增長額節點。

步驟三：配置資料推送節點

通過節點上下文參數的本節點輸入參數接收上遊查詢昨日銷售總額與查詢昨日銷售增長額的輸出參數outputs，從而可以在本文中使用參數並將其推送至目標。

1. 雙擊名為推送昨日銷售總額與增長額的資料推送節點，進入節點後，單擊調度配置，在調度配置面板中進行如下配置。

   參數類型		參數內容	圖示
   調度參數	參數名	curdate
   	參數值	$[yyyymmddhh:mi:ss]
   時間屬性	調度周期	日
   	定時調度時間	08：00 說明 本實踐以8:00為例，確保資料能在08:00時推送至目標渠道。您可以根據實際需要配置其他時間點。
   	重跑屬性	運行成功或失敗後皆可重跑。
   資源屬性	調度資源群組	選擇已建立好的調度資源群組。 說明 初次使用資料推送節點，需先提工單升級調度資源群組。
   節點上下文參數	本節點輸入參數	單擊添加新增本節點輸入參數： 參數一： 參數名：inputs1。 取值來源：選擇上遊節點查詢昨日銷售總額的輸出參數outputs。 參數二： 參數名：inputs2。 取值來源：選擇上遊節點查詢昨日銷售增長額的輸出參數outputs。

2. 調度配置完成後，即可配置資料推送內容，詳情如下。

   • 資料推送目標：下拉資料推送目標選擇所需的資料推送目標，若不存在，可單擊下拉框右下角的建立資料推送目標，建立推送目標。

   參數	說明
   類型	支援圈選DingTalk、飛書、企業微信、Teams以及郵件。
   對象名稱	可按業務需求進行自訂。
   WebHook	DingTalk、飛書、企業微信機器人或Teams的Webhook，以及郵件的SMTP需要在相應的目標平台上擷取。 說明 飛書機器人Webhook擷取方式可參見：飛書Webhook觸發器配置。 Teams Webhook擷取方式可參見：使用 Microsoft Teams 工作流程建立傳入 Webhook。

   • 標題：推送昨日銷售總額與增長額

   • 本文：按需求進行配置本文即可，詳情可參見配置推送內容。

     說明

     本文中可直接使用上遊SQL查詢節點查詢的欄位名作為預留位置，以擷取上遊輸入的參數。

3. 配置完成後，單擊儲存推送昨日銷售總額與增長額推送節點。

步驟四：測試合并推送流程

在完成簡單推送流程配置後，才可對合并推送測試流程進行測試，以方便後續提交發布。

1. 雙擊開啟資料推送Demo流程商務程序圖頁面。

2. 選中推送昨日銷售總額與增長額節點，右鍵選擇運行到該節點，等待運行完成即可。

   說明

   如果出現任務運行失敗，選中需要查看的節點，單擊右鍵選擇查看日誌即可查看日誌。

   

步驟一：搭建推送流程

雙擊資料推送Demo流程後，在商務程序圖頁面，可以看到已建立的多個節點，需拖動離線同步節點作為賦值節點的上遊。將MySQL同步資料至ODPS、ODPS資料查詢以及ODPS資料推送三個節點依次串連起來。

步驟二：配置離線同步節點

通過配置MySQL同步資料至ODPS節點，實現將準備資料階段在MySQL中寫入的測試資料同步至ODPS資料來源，以供後續使用。

1. 雙擊MySQL同步資料至ODPS節點進入離線同步節點的配置。

   配置項		配置內容	圖示
   資料來源側	資料來源	MySQL
   	資料來源名稱	選擇您建立的MySQL資料來源。
   我的資源群組		選擇Serverless 資源群組。
   資料去向側	資料去向	MaxCompute（ODPS）
   	資料來源名稱	選擇該空間綁定的MaxCompute資料來源。

   配置完成後，系統會自動進行資料來源與資源群組之間的連通性測試，待連通性展示為成功後即可單擊下一步，進行資料來源與去向的詳細配置。

2. 配置資料來源與去向。

   配置項		配置內容	圖示
   資料來源	資料來源	保持預設： MySQL。 選擇您建立的MySQL資料來源。
   	表	選擇orders表。
   	資料過濾	可按需求進行配置，本實踐中置空即可。
   	切分鍵	您可以將來源資料表中某一列作為切分鍵，建議使用主鍵或有索引的列作為切分鍵。
   	資料預覽	預覽從MySQL資料來源中擷取到的資料，可用於判斷是否符合預期。
   資料去向	資料來源	保持預設： MaxCompute（ODPS）。 選擇該空間綁定的MaxCompute資料來源。
   	Tunnel資源群組	即 Tunnel Quota，預設選擇“公用傳輸資源”，即MC的免費quota。
   	表	單擊一鍵產生目標表結構，產生目標表。
   	分區資訊	如果您每日增量資料限定在對應日期的分區中，可以使用分區做每日增量，比如配置分區pt值為${bizdate}
   	寫入模式	寫入前清理已有資料（Insert Overwrite）

3. 配置好來源與去向後，將欄位進行同名映射即可。

4. 通道配置。

   • 任務期望最大並發數：由於資源原因或者任務本身特性等原因，實際執行時並發數可能小於等於此值，收費按照實際執行的並發數收費，本實踐選擇2個並發即可。

   • 同步速率：通過限流可以保護資料來源端或者資料去向端的讀寫壓力，不限流的情況下則會提供現有硬體環境下最大的傳輸效能，本實踐不進行限流。

   • 髒資料策略：不容忍髒資料。

   • 分散式處理能力：預設為關閉狀態，需要並發數大於等於8才能開啟分布式。

   

5. 調度配置。

   完成整合任務配置後，單擊右側調度配置，在彈出的調度面板中配置如下內容。

   • 調度參數：

     • 參數名：bizdate

     • 參數值：$[yyyymmdd-1]

   • 定時調度時間：08：00

   • 重跑屬性：下拉選擇運行成功或失敗後皆可重跑。

   • 調度資源群組：選擇已建立好的調度資源群組。

   • 調度依賴：勾選使用工作空間根節點，作為離線同步任務的上遊節點。

6. 單擊儲存MySQL同步資料至ODPS離線同步節點。

步驟三：配置賦值節點

ODPS資料來源不支援使用ODPS SQL節點進行查詢並通過節點上下文參數輸出資料至資料推送節點。需採用賦值節點對ODPS資料進行查詢處理，再通過節點上下文參數進行輸出至資料推送節點。

1. 雙擊ODPS資料查詢節點，進入賦值節點。

   1. 在賦值節點編輯頁面上方的請選擇賦值語言的下拉框裡選擇ODPS SQL。

   2. 寫入ODPS SQL。

      -- 首先，我們建立一個帶有排名的子查詢，用於計算每個分區（pt）內按銷售額（sales）降序排列的順序。
      -- DENSE_RANK() 函數會為每個分區內的行提供一個唯一的排名，相等的銷售額會得到相同的排名，但不會跳過任何一個排名數字。
      --
      -- 子查詢部分：
      -- 1. 從'orders'表中選擇需要的列：order_id（訂單ID）、category（類別）、sales（銷售額）、datetime（時間）和pt（業務日期）。
      -- 2. 使用DENSE_RANK() OVER (PARTITION BY pt ORDER BY sales DESC)：
         -- PARTITION BY pt 指定按照pt欄位進行分區，即對每個不同的pt值分別進行排名。
         -- ORDER BY sales DESC 表示在每個分區內按照sales欄位的降序來排序，即銷售額最高的排在前面。
         -- rank 列將會儲存每行在各自分區內的銷售排名。
      --
      -- 主查詢部分：
      -- 從上述帶有排名的子查詢中篩選出排名在前三名（rank <= 3）的記錄。
      -- 這意味著對於每個業務日期，我們只選取銷售額最高的前三個訂單的資訊。
      
      SELECT
        order_id,          -- 選擇訂單ID
        category,          -- 選擇商品類別
        sales,             -- 選擇銷售額
        datetime,          -- 選擇訂單時間
        pt                  -- 選擇業務日期
      FROM (
          SELECT
            order_id,
            category,
            sales,
            datetime,
            pt,
            DENSE_RANK() OVER (PARTITION BY pt ORDER BY sales DESC) AS rank  -- 計算每個pt分區內的銷售額排名
          FROM orders
          WHERE pt = '${bizdate}'  -- 篩選業務日期為${bizdate}的記錄
      ) AS ranked_orders
      WHERE rank <= 3  -- 只保留每個pt分區銷售額排名前三的記錄

2. 編輯完成SQL代碼後，單擊右側調度配置，在彈出的調度配置面板中配置如下內容。

   • 定時調度時間：08：00。

   • 調度資源群組：選擇已建立好的Serverless資源群組。

   • 依賴上遊節點：檢查依賴的上遊是否為MySQL同步資料至ODPS的離線同步節點。

   • 節點上下文參數：單擊本節點輸出參數後的添加，添加輸出參數作為下遊節點輸入參數的取值。

     

3. 單擊儲存ODPS資料查詢節點。

步驟四：配置資料推送節點

通過配置節點上下文參數的本節點輸入參數，來擷取上遊賦值節點輸出的outputs參數，在本文中使用這些參數並將其推送至目標。

1. 雙擊名為ODPS資料推送的資料推送節點，進入節點後，單擊調度配置，在調度配置面板中進行如下配置。

   參數類型		參數內容	圖示
   調度參數	參數名	curdate
   	參數值	$[yyyymmddhh:mi:ss]
   時間屬性	調度周期	日
   	定時調度時間	08：00 說明 本實踐以8:00為例，確保資料能在08:00時推送至目標渠道。您可以根據實際需要配置其他時間點。
   	重跑屬性	運行成功或失敗後皆可重跑。
   資源屬性	調度資源群組	選擇資料推送節點功能上線日期後建立的資源群組，若資源群組為發布日期前，需提工單升級調度資源群組。 說明 資料推送節點發布日期為2024年6月28日，DataWorks更多發布記錄可參見：功能發布記錄。
   節點上下文參數	本節點輸入參數	單擊添加新增本節點輸入參數： 參數名：inputs 取值來源：選擇上遊節點ODPS資料查詢的輸出參數outputs。

2. 調度配置完成後，即可配置資料推送內容，詳情如下。

   • 資料推送目標：下拉資料推送目標選擇所需的資料推送目標，若不存在，可單擊下拉框右下角的建立資料推送目標，建立推送目標。

   參數	說明
   類型	支援圈選DingTalk、飛書、企業微信、Teams以及郵件。
   對象名稱	可按業務需求進行自訂。
   WebHook	DingTalk、飛書、企業微信機器人或Teams的Webhook，以及郵件的SMTP需要在相應的目標平台上擷取。 說明 飛書機器人Webhook擷取方式可參見：飛書Webhook觸發器配置。 Teams Webhook擷取方式可參見：使用 Microsoft Teams 工作流程建立傳入 Webhook。

   • 標題：ODPS資料推送

   • 本文：按需求進行配置即可，詳情可參見配置推送內容。

     說明

     本文中可直接使用上遊賦值節點查詢的欄位名作為預留位置，以擷取上遊輸入的參數。

3. 配置完成後，單擊儲存ODPS資料推送節點。

步驟五：測試MaxCompute資料推送流程

在完成資料推送流程節點配置後，需要對資料推送流程進行測試，以便後續提交發布。

1. 雙擊開啟資料推送Demo流程商務程序頁面。

2. 選中MySQL同步資料至ODPS節點，右鍵選擇運行節點及下遊，等待運行完成即可。

   說明

   如果出現任務運行失敗，選中需要查看的節點，單擊右鍵選擇查看日誌即可查看日誌。