Elasticsearch：通過DataWorks將MaxCompute資料同步到阿里雲ES

步驟一：準備來源資料

建立MaxCompute表並匯入測試資料。具體操作，請參見建立表和匯入資料。

本文使用的表結構和表資料如下所示：

• 表結構

  表包含 7 個欄位和 1 個分區欄位，欄位定義如下：

  • create_time（string，主鍵）

  • category（string）

  • brand（string）

  • buyer_id（string）

  • trans_num（bigint）

  • trans_amount（double）

  • click_cnt（bigint）

  分區欄位為 pt（bigint）。

• 部分表資料

  來源資料表包含以下欄位：

  • create_time：交易日期，如 2020/6/1

  • category：商品類目，如外套、生鮮、電器、衛浴

  • brand：品牌名稱，如品牌 A～品牌 G

  • buyer_id：買家 ID，如 user1～user13

  • trans_num：交易數量

  • trans_amount：交易金額

  • click_cnt：點擊次數

  • pt：分區欄位，值為 1

步驟二：購買並配置獨享資源群組

購買一個Data Integration獨享資源群組，並為該資源群組綁定專用網路和工作空間。獨享資源群組可以保證資料快速、穩定地傳輸。

1. 登入DataWorks控制台。

2. 在頂部功能表列選擇相應地區後，在左側導覽列單擊資源群組。

3. 在獨享資源群組頁簽下，單擊建立舊版資源群組 > 建立舊版資源群組 > Data Integration資源群組。

4. 在DataWorks獨享資源（訂用帳戶）購買頁面，獨享資源類型選擇獨享Data Integration資源，輸入資源群組名稱，單擊立即購買，購買獨享資源群組。

   更多配置資訊，請參見步驟一：購買資源群組。

5. 在已建立的獨享資源群組的操作列，單擊網絡設定，為該獨享資源群組綁定專用網路。具體操作，請參見綁定專用網路。

   說明

   本文以獨享Data Integration資源群組通過VPC內網同步資料為例。關於通過公網同步資料，請參見添加白名單。

   獨享資源需要與Elasticsearch執行個體的專用網路連通才能同步資料。因此需要綁定Elasticsearch執行個體所在的专有网络、可用区和交換器。查看Elasticsearch執行個體所在的專用網路、可用性區域和交換器，請參見查看Elasticsearch執行個體的基本資料。

   重要

   綁定專用網路後，您需要將專用網路的交換機網段加入到Elasticsearch執行個體的VPC私網訪問白名單中。具體操作，請參見配置Elastic search執行個體公網或私網訪問白名單。

6. 在頁面左上方，單擊返回表徵圖，返回资源组列表頁面。

7. 在已建立的獨享資源群組的操作列，單擊綁定工作空間，為該獨享資源群組繫結目標工作空間。

   具體操作，請參見步驟二：綁定歸屬工作空間。

步驟三：添加資料來源

將MaxCompute和Elasticsearch資料來源接入DataWorks的Data Integration服務中。

1. 進入DataWorks的快速進入 > Data Integration頁面。

   1. 登入DataWorks控制台。

   2. 在左側導覽列，單擊工作空間。

   3. 在目標工作空間的操作列，選擇快速進入 > 快速進入 > Data Integration。

2. 在左側導覽列，單擊資料來源。

3. 新增MaxCompute資料來源。

   1. 在資料來源列表頁面，單擊新增資料來源。

   2. 在新增資料來源頁面，搜尋並選擇MaxCompute資料來源。

   3. 在新增MaxCompute資料來源對話方塊，在基礎資訊地區配置資料來源參數。

      配置詳情，請參見配置MaxCompute資料來源。

   4. 在連接配置地區，單擊測試連通性，連通狀態顯示為可連通時，表示連通成功。

   5. 單擊完成。

4. 使用同樣的方式添加Elasticsearch資料來源。配置詳情，請參見配置Elasticsearch資料來源。

步驟四：配置並運行資料同步任務

資料同步任務將獨享資源群組作為一個可以執行任務的資源，獨享資源群組將擷取Data Integration服務中資料來源的資料，並將資料寫入Elasticsearch。

1. 進入DataWorks的快速進入 > Data Integration。 在左側導覽列，單擊資料來源頁面。

   1. 登入DataWorks控制台。

   2. 在左側導覽列，單擊工作空間。

   3. 在目標工作空間的操作列，選擇快速進入 > 快速進入 > Data Integration。 在左側導覽列，單擊資料來源。

2. 建立一個離線同步任務。

   1. 在左側導覽列的資料開發（表徵圖）頁簽，選擇 建立 > 建立業務流，按照介面指引建立一個商務程序。

   2. 按右鍵建立的商務程序，選擇建立節點 > 建立節點 > Data Integration > 離線同步 > 到阿里雲ES中。

   3. 在建立節點對話方塊中，輸入節點名稱，單擊確認。

3. 配置網路與資源

   1. 在資料來源區域，資料來源選擇MaxCompute(ODPS)，資料來源名稱選擇待同步的資料來源名稱。 在我的資源群組地區，資料來源區域，資料來源選擇MaxCompute(ODPS)，資料來源名稱選擇待同步的資料來源名稱。 在我的資源群組選擇MaxCompute(ODPS)，資料來源名稱選擇待同步的資料來源名稱。

   2. 在我的資源群組地區，選擇獨享資源群組。

   3. 在資料去向地區，資料去向選擇Elasticsearch，資料來源名稱選擇待同步的資料來源名稱。

4. 單擊下一步。

5. 配置任務。

   1. 在資料來源區域，資料來源選擇MaxCompute(ODPS)，資料來源名稱選擇待同步的資料來源名稱。 在我的資源群組地區，選擇待同步的表。

   2. 在資料去向地區，配置資料去向的各參數。

   3. 在欄位對應地區中，設定來源欄位與目標欄位的映射關係。

   4. 在通道控制地區，配置通道參數。

   詳細配置資訊，請參見嚮導模式配置。

6. 運行任務。

   1. （可選）配置任務調度屬性。在頁面右側，單擊調度配置，按照需求配置相應的調度參數。各配置的詳細說明，請參見調度配置。

   2. 在節點地區的右上方，單擊儲存表徵圖，儲存任務。

   3. 在節點地區的右上方，單擊提交表徵圖，提交任務。

      如果您配置了任務調度屬性，任務會定期自動執行。您還可以在節點地區的右上方，單擊運行表徵圖，立即運行任務。

      作業記錄中出現Shell run successfully!表明任務運行成功。部分任務作業記錄如下所示：

      2023-10-31 16:52:35 INFO Exit code of the Shell command 0
      2023-10-31 16:52:35 INFO --- Invocation of Shell command completed ---
      2023-10-31 16:52:35 INFO Shell run successfully!
      2023-10-31 16:52:35 INFO Current task status: FINISH
      2023-10-31 16:52:35 INFO Cost time is: 33.106s

步驟五：驗證資料同步結果

在Kibana控制台中，查看同步成功的資料，並按條件查詢資料。

1. 登入目標Elasticsearch執行個體的Kibana控制台。

   具體操作，請參見登入Kibana控制台。

2. 單擊Kibana頁面左上方的表徵圖，選擇Dev Tools（開發工具）。

3. 在Console（控制台）中，執行如下命令查看同步的資料。

   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {}}
   }

   說明

   odps_index為您在資料同步指令碼中設定的index欄位的值。

   資料同步成功後，返回如下結果。

   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {}}
   }
   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "_source": ["category", "brand"]
   }
   POST /odps_index/_search?pretty
   {
     "query": { "match": {"category":"生鮮"} }
   }
   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "sort": { "trans_num": { "order": "desc" }
           }
   }
   
   --- Response ---
   {
     "took" : 2,
     "timed_out" : false,
     "_shards" : {
       "total" : 1,
       "successful" : 1,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : 13,
       "max_score" : null,
       "hits" : [
         {
           "_index" : "odps_index",
           "_type" : "_doc",
           "_id" : "2020/6/7 8:00",
           "_score" : null,
           "_source" : {
             "trans_num" : 88,
             "click_cnt" : 80,
             "category" : "外套",
             "buyer_id" : "user7",
             "trans_amount" : 150.0,
             "brand" : "品牌E"
           },
           "sort" : [
             88
           ]
         },
         {
           "_index" : "odps_index",
           "_type" : "_doc",
           "_id" : "2020/6/11 8:00",
           "_score" : null,
           "_source" : {
             "trans_num" : 22,
             "click_cnt" : 70,
             "category" : "衛浴",
             "buyer_id" : "user11",
             "trans_amount" : 4500.0,
             "brand" : "品牌G"
           },
           "sort" : [
             22
           ]
         }
       ]
     }
   }

4. 執行如下命令，搜尋文檔中的category和brand欄位。

   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "_source": ["category", "brand"]
   }

5. 執行如下命令，搜尋category為生鮮的文檔。

   POST /odps_index/_search?pretty
   {
   "query": { "match": {"category":"生鮮"} }
   }

6. 執行如下命令，按照trans_num欄位對文檔進行排序。

   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "sort": { "trans_num": { "order": "desc" } }
   }

   更多命令和訪問方式，請參見Elastic.co官方協助中心。