Dataphin：跨節點參數最佳實務

應用情境

跨節點參數有兩個典型的應用情境：

• 某金融企業每天的周期任務涉及到幣種轉換，需要將每日的投資收益按照美元進行結算。每日的收盤匯率會從業務系統中提取並同步至一個離線表。多個周期任務均涉及幣種換算，這需要使用業務日期相應的匯率。可以將讀取匯率的程式編寫為一個輸出節點任務，將匯率作為跨節點參數輸出，需要匯率的節點任務則作為輸入節點來引用該跨節點參數。

• 某經銷零售企業，需要每天定時查詢供應商的某個服務，以確認產品目錄是否有變更，如有變更，則需要全量拉取最新的目錄資料（由於資料量較大，拉取一次的成本較高）；若未發生變更，則繼續使用上一次拉取的資料。可以將檢測產品目錄是否變更的程式寫入至一個輸出節點任務中，將產品目錄的更新狀態作為跨節點參數輸出，同時將拉取並同步產品目錄的節點設為輸入節點，根據跨節點參數的取值進行調度（條件調度）。

使用樣本一：匯率轉換

資料結構

匯率表exchange_rate_table，表結構和樣本資料如下。

輸出節點

1. 建立MAX_COMPUTE_SQL任務get_exchange_rate，並定義跨節點輸出參數usd2cny。

   

2. 在任務編輯器中，單擊滑鼠右鍵，選擇設定跨節點參數。

3. 任務編輯器將自動提示已聲明定義好的跨節點輸出參數。

   

4. 將某一欄位的別名設定為跨節點輸出參數的名稱，系統將自動將查詢結果第一行相應欄位的取值賦值給跨節點參數。

   說明

   對於SQL任務，如果有多個語句輸出跨節點參數，則每一個語句前的set語句不可省略。

   本例有兩種SQL寫法：

   • 每一個跨節點參數一個獨立的 SQL。

     

   • 使用一個語句。

     

5. 提交任務。

輸入節點

1. 建立SQL任務exchange_usd_to_cny，範例程式碼如下。同時，將輸出節點get_exchange_rate添加為上遊。

   -- 10000 美金換算為人民幣，日元，歐元，澳幣和港幣
   select 10000 * ${usd2cny_rate}, 
    10000 * ${usd2jpy_rate}, 
    10000 * ${usd2eur_rate}, 
    10000 * ${usd2aud_rate}, 
    10000 * ${usd2hkd_rate};

   

2. 將識別出的變數參數的類型修改為跨節點變數。

   

3. 在各個跨節點變數的參數值中選擇get_exchange_rate的相應跨節點輸出參數

   

補資料

1. 確認表中當日的匯率。

   

2. 進入營運中心，在當前任務的get_exchange_rate節點中單擊補資料，選擇補當前及上下遊任務，選中下遊exchange_usd_to_cny一起補資料。

   說明

   對輸入節點補資料時，必須連帶補輸出節點，否則輸入節點將使用跨節點參數的預設值。

   

3. exchange_usd_to_cny的作業記錄如下，可以看到系統從表中將資料讀取出來賦值給跨節點輸出參數，並傳遞給了下遊任務。

   

樣本二：確認更新狀態

輸出節點

1. 建立一個類比的檢測更新狀態的Python任務，並添加跨節點輸出參數update_status。

   

2. 輸入代碼，使用隨機函數來返回狀態。單擊滑鼠右鍵，選擇設定跨節點參數後，提交任務。

   

   

   from random import randint
   def check_update():
    return randint(0, 1)
   setv("update_status", check_update())

輸入節點

1. 建立一個離線管道任務imp_product_catalog，並將check_update作為其上遊任務添加。

   

2. 為imp_product_catalog任務開啟條件調度。

   

3. 條件調度中添加條件跨節點參數-check_update.update_status = 0 （即無更新）時，空跑調度，不符合該條件則命中預設條件（即正常調度）。

   

   

   

4. 輸入不同的跨節點參數值，預覽調度運行計劃。