通過DataWorks操作Hologres的多個分區子表

背景資訊

Hologres相容原生PostgreSQL生態，支援分區表，在對分區表執行INSERT、DELETE或UPDATE操作時，需要對子表執行，不能在父表中操作然後自動路由到子表。因此對於分區表的操作通常建議使用DataWorks調度節點周期性執行，例如通過DataWorks周期性匯入MaxCompute資料最佳實務。

但是對於回刷歷史資料等一次性涉及多個分區子表操作的情境，調度需要執行多個SQL較為麻煩，因此本文將會介紹如何使用DataWorks的for-each節點實現多分區操作。

使用限制

您需要購買DataWorks標準版及以上版本，才可以在Hologres SQL節點中使用賦值參數功能。

使用樣本

本文以MaxCompute資料寫入Hologres多個分區子表為例，通過for-each節點實現多分區操作的流程如下：

1. 準備表和資料。

   1. 在MaxCompute中建立MaxCompute分區表，並插入資料。

      --建立一張MaxCompute分區表sale_detail。
      CREATE TABLE IF NOT EXISTS odps_sale_detail (
          shop_name STRING,
      	  customer_id STRING,
          total_price DOUBLE
      )
      PARTITIONED BY ( sale_date STRING);
      ---- 向源表增加分區20240110並寫入資料
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20240110');
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20240110') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3);
      -- 向源表增加分區20240111
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20240111');
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20240111') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3);
      -- 向源表增加分區20240112並寫入資料
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20240112');
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20240112') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3);
      -- 向源表增加分區20240113並寫入資料
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20240113');
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20240113') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3);
      -- 向源表增加分區20240114並寫入資料
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20240114');
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20240114') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3);
      -- 向源表增加分區20240115並寫入資料
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20240115');
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20240115') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3);
      -- 向源表增加分區20240116並寫入資料
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20240116');
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20240116') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3);

   2. 在Hologres中建立外部表格和分區父表。

      -- 建立外部表格
      IMPORT FOREIGN SCHEMA <maxCompute_projectname> LIMIT to(
        odps_sale_detail) 
      FROM SERVER odps_server INTO public 
      OPTIONS(if_table_exist 'error',if_unsupported_type 'error');
      -- 建立Hologres分區父表（內部表）
      BEGIN ;
      CREATE TABLE IF NOT EXISTS holo_sale_detail(
          shop_name TEXT,
      	  customer_id TEXT ,
          total_price FLOAT8,
          sale_date TEXT
      )
      PARTITION BY LIST(sale_date);
      COMMIT;

2. DataWorks工作空間綁定Hologres資料來源。

   需要在DataWorks中綁定Hologres資料來源，詳情請參見綁定Hologres計算資源。

3. 添加Hologres SQL節點擷取分區。

   1. 在DataWorks数据开发模組建立Hologres SQL節點，輸入相應的SQL以擷取分區值，作為for-each節點的輸入。建立Hologres SQL節點的詳情請參見Hologres SQL節點。

      樣本SQL：擷取20240112~20240116共計5個分區值。

      SELECT distinct sale_date FROM odps_sale_detail WHERE sale_date > '20240111';

      返回結果：

      +------------+
      | sale_date  |
      +------------+
      | 20240112   |
      | 20240113   |
      | 20240114   |
      | 20240115   |
      | 20240116   |
      +------------+

   2. 為該節點設定輸出參數。

      在調度配置 > 節點上下文參數 > 本節點輸出參數地區單擊添加賦值參數，選擇預設的outputs參數即可。詳情請參見配置節點上下文。系統自動添加一條輸出參數：參數名為 outputs，取值為 ${outputs}，表示本節點的最後一個輸出結果。

4. 配置for-each節點，迴圈操作分區資料，具體操作可參考配置for-each節點。

   1. 添加for-each節點。

   2. 在for-each節點內新增一個Hologres SQL節點，設定其上遊為start節點，下遊為end節點，並在Hologres SQL節點中輸入對應的資料匯入邏輯。樣本SQL如下：

      說明

      您需要先建立對應的子表，並插入資料。for-each節點會將上一步中Hologres SQL節點的分區結果賦值給指令碼中的參數，然後迴圈執行，直到輸出值全部匹配完成。

      --建子表
      CREATE TABLE IF NOT EXISTS  public.holo_sale_detail_${dag.foreach.current} PARTITION OF public.holo_sale_detail FOR VALUES IN('${dag.foreach.current}');
      --匯入資料
      INSERT  INTO public.holo_sale_detail_${dag.foreach.current} SELECT * FROM odps_sale_detail where sale_date ='${dag.foreach.current}';

   3. 在for-each節點的編輯頁面，單擊右側的調度配置，在調度依賴地區配置本節點依賴的上遊節點。並在節點上下文 > 本節點輸入參數地區單擊預設參數名loopDataArray後的編輯。從取值來源列表中，選擇上遊賦值節點的outputs參數。在 for-each-節點的節點上下文參數配置中，本節點輸入參數中已包含參數 loopDataArray，其取值來源引用了上遊節點的 t.outputs，描述為"賦值參數，取值為本節點的最後一個輸出結果"。

   4. 儲存並提交for-each節點。

      說明

      如果您使用的是標準模式的工作空間，提交成功後，請單擊右上方的發佈。具體操作請參見發布任務。

5. 測試並查看測試結果，具體操作可參見測試並查看測試結果。

   1. 在周期任务頁面相應節點的右側的DAG圖中，按右鍵賦值節點（即第一個節點），選中补数据 > 當前節點及下游節點。

   2. 待補資料執行個體運行成功後，在補數據實例頁面，按右鍵遍曆節點（即第二個節點），選中查看內部節點，即可看到作業中迴圈了5次執行。補資料觸發後，在運行執行個體頁面可查看 for-each 迴圈節點的執行情況：左側迭代列表顯示共 5 次迭代且全部運行成功，右側 DAG 圖展示每次迭代的工作流程為 for-each Start → Hologres SQL → for-each End，三個節點均執行成功。頁面頂部提示迴圈節點最大迴圈次數為 128 次，超出後將直接運行失敗。

   3. 在Hologres中可查看新的子表已建立成功並寫入了資料。補資料運行成功後，在HoloWeb中可看到 public schema 下 holo_sale_detail 父表已產生 5 個子分區表（holo_sale_detail_20240112 至 holo_sale_detail_20240116）。執行 SELECT * FROM public.holo_sale_detail_20240113; 查詢指定分區，結果返回 shop_name、customer_id、total_price、sale_date 四列資料，驗證各分區資料寫入正確。

失敗處理

如果在執行for-each過程中，某一次迴圈出錯，則剩餘任務都會停止運行。您可以前往for-each節點內部查看失敗詳情。當for-each節點執行失敗時，單擊該節點並開啟作業記錄Tab頁簽。若提示"因當前節點存在內部節點，無法直接查看作業記錄"，需單擊提示中的連結，先進入內部節點，再查看內部節點的作業記錄。

失敗處理完成後，您可以對任務執行重跑，即可重新運行失敗任務。按右鍵失敗的 for-each-節點，在彈出的操作功能表中單擊重跑，即可重新運行該節點。