E-MapReduce：基於Flink、EMR Serverless Spark與Paimon構建流批一體資料湖分析

Realtime ComputeFlink版

阿里雲Realtime ComputeFlink版是一種全託管Serverless的Flink雲端服務，是一站式開發營運管理平台，開箱即用，計費靈活。具備作業開發、資料調試、運行與監控、自動調優、智能診斷等全生命週期能力。更多資訊，請參見什麼是阿里雲Realtime ComputeFlink版。

Apache Paimon

Apache Paimon是一種統一的資料湖儲存格式，結合Flink和Spark構建了流批處理的即時湖倉一體架構。Paimon創新地將湖格式與LSM（Log-structured merge-tree）技術結合，使資料湖具備了即時資料流更新和完整的流處理能力。更多資訊，請參見Apache Paimon。

步驟一：通過Realtime ComputeFlink建立Paimon Catalog

Paimon Catalog可以方便地管理同一個warehouse目錄下的所有Paimon表，並與其它阿里雲產品連通。建立並使用Paimon Catalog，詳情請參見管理Paimon Catalog。

1. 登入Realtime Compute控制台。

2. 單擊目標工作空間操作列下的控制台。

3. 建立Paimon Catalog。

   通過UI方法添加DLF類型的Paimon Catalog，無需手動填寫AccessKey資訊。具體操作，請參見管理Paimon Catalog。

4. 建立Paimon表。

   在查询脚本文本編輯地區輸入如下命令後，選中代碼後單擊运行。

   CREATE TABLE IF NOT EXISTS `paimon`.`test_paimon_db`.`test_append_tbl`
   (
       id       STRING,
       data     STRING,
       category INT,
       ts       STRING,
       dt       STRING,
       hh       STRING
   ) PARTITIONED BY (dt, hh)
   WITH (
       'write-only' = 'true'
   );

5. 建立流作業。

   1. 新增作業。

      1. 在左側導覽列，選擇数据开发 > ETL。

      2. 建立流作業，在新增作業草稿對話方塊中，填寫作業配置資訊。

         作業參數	說明
         文件名称	作業的名稱。 說明 作業名稱在當前專案中必須保持唯一。
         引擎版本	當前作業使用的Flink的引擎版本。引擎版本號碼含義、版本對應關係和生命週期重要時間點詳情請參見引擎版本介紹。

      3. 單擊创建。

   2. 編寫代碼。

      在建立的作業草稿中，輸入以下代碼，通過datagen源源不斷產生資料寫入Paimon表中。

      CREATE TEMPORARY TABLE datagen
      (
          id        string,
          data      string,
          category  int
      )
      WITH (
          'connector' = 'datagen',
          'rows-per-second' = '100',
          'fields.category.kind' = 'random',
          'fields.category.min' = '1',
          'fields.category.max' = '10'
      );
      INSERT INTO `paimon`.`test_paimon_db`.`test_append_tbl`
      SELECT
          id,
          data,
          category,
          cast(LOCALTIMESTAMP as string) as ts,
          cast(CURRENT_DATE as string) as dt,
          cast(hour(LOCALTIMESTAMP) as string) as hh
      FROM datagen;

   3. 單擊部署，即可將資料發布至生產環境。

   4. 您可以在作業營運頁面啟動作業進入運行階段，詳情請參見作業啟動。

步驟二：通過EMR Serverless Spark建立SQL會話

建立的SQL會話用於SQL開發和查詢。有關會話的詳細介紹，請參見會話管理。

1. 進入會話管理頁面。

   1. 登入E-MapReduce控制台。

   2. 在左側導覽列，選擇EMR Serverless > Spark。

   3. 在Spark頁面，單擊目標工作空間名稱。

   4. 在EMR Serverless Spark頁面，單擊左側導覽列中的會話管理。

2. 在Serverless Spark工作空間的資料目錄中，添加步驟一中建立的DLF Catalog。

3. 建立SQL會話。

   1. 在SQL 会话頁簽，單擊创建 SQL 会话。

   2. 在建立SQL會話頁面，配置以下資訊，其餘參數無需配置，然後單擊创建。

      參數	說明
      名称	自訂SQL會話的名稱。例如，paimon_compute。

   3. 單擊操作列的启动。

步驟三：通過EMR Serverless Spark進行互動式查詢或任務調度

EMR Serverless Spark提供了互動式查詢和任務調度兩種操作模式，以滿足不同的使用需求。互動式查詢適用於快速查詢和調試，而任務調度則支援任務的開發、發布和營運，實現完整的生命週期管理。

在資料寫入過程中，我們可以隨時使用EMR Serverless Spark對Paimon表進行互動式查詢，以便即時擷取資料狀態和執行快速分析。此外，通過發布開發好的任務並建立工作流程，可以編排各項任務並完成工作流程的發布。您可以配置調度策略，實現任務的定期調度，從而保證資料處理和分析的自動化與高效性。