paimon準系統 - Realtime Compute for Apache Flink

本文通過簡單的樣本，在Realtime Compute開發控制台帶您快速體驗Paimon的準系統，包括建立和刪除Paimon Catalog，建立和刪除Paimon Catalog表，向Paimon表寫入、更新以及消費資料。

前提條件

如果您使用RAM使用者或RAM角色等身份訪問，需要確認已具有Realtime Compute開發控制台相關許可權，詳情請參見許可權管理。
已建立Flink工作空間，詳情請參見開通Realtime ComputeFlink版。

已開通Object Storage Service且Bucket的儲存類型為標準儲存，詳情請參見OSS控制台快速入門。OSS用於儲存Paimon表的相關檔案，包括資料檔案與中繼資料檔案等。
僅Realtime Compute引擎VVR 8.0.5及以上版本支援Paimon表。

步驟一：建立Paimon Catalog

進入查詢指令碼頁面。
1. 登入Realtime Compute控制台。
2. 單擊目標工作空間操作列下的控制台。
3. 在左側導覽列，單擊資料開發 > 資料查詢，在查詢指令碼頁簽，建立查詢指令碼。

在文本編輯地區，輸入如下代碼建立Paimon Catalog。

-- my-Catalog為自訂的Catalag名稱
CREATE Catalog `my-catalog` WITH (
  'type' = 'paimon',
  'metastore' = 'filesystem',
  'warehouse' = '<warehouse>',
  'fs.oss.endpoint' = '<fs.oss.endpoint>',
  'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
  'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
);

參數配置項如下：

配置項	說明	是否必填	備忘
type	Catalog類型。	是	固定值為Paimon。
metastore	中繼資料存放區類型。	是	本文樣本填寫filesystem，其他類型詳情請參考管理Paimon Catalog。
warehouse	OSS服務中所指定的數倉目錄。	是	格式為oss://<bucket>/<object>。其中： bucket：表示您建立的OSS Bucket名稱。 object：表示您存放資料的路徑。請在OSS管理主控台上查看您的Bucket和Object名稱。
fs.oss.endpoint	OSS服務的串連地址。	否	當warehouse指定的OSS Bucket與Flink工作空間不在同一地區，或使用其它帳號下的OSS bucket時需要填寫。請參見訪問網域名稱和資料中心。說明如果需要將Paimon表格儲存體在OSS-HDFS中，則fs.oss.endpoint、fs.oss.accessKeyId與fs.oss.accessKeySecret均需要填寫，其中fs.oss.endpoint的值為cn-<region>.oss-dls.aliyuncs.com，例如cn-hangzhou.oss-dls.aliyuncs.com。
fs.oss.accessKeyId	擁有讀寫OSS許可權的阿里雲帳號或RAM帳號的AccessKey。	否	當warehouse指定的OSS Bucket與Flink工作空間不在同一地區，或使用其它帳號下的OSS Bucket時需要填寫。擷取方法請參見建立AccessKey。
fs.oss.accessKeySecret	擁有讀寫OSS許可權的阿里雲帳號或RAM帳號的AccessKey Secret。	否

選中建立Paimon Catalog代碼，單擊左側的運行。
返回The following statement has been executed successfully!資訊表示Catalog建立成功。

步驟二：建立Paimon Catalog表

在查詢指令碼頁面，輸入如下代碼建立名為my_db的Paimon資料庫以及名為my_tbl的Paimon表。
```
CREATE DATABASE `my-catalog`.`my_db`;
CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  dt STRING,
  id BIGINT,
  content STRING,
  PRIMARY KEY (dt, id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'changelog-producer' = 'lookup'  
);
```
說明
為了後續能夠流式消費Paimon表，本樣本在WITH中指定了changelog-producer參數為lookup，表示採用尋找策略產生變更日誌，詳情請參見變更資料產生機制。
選中建立Paimon資料庫以及Paimon表的代碼，單擊左側的運行。
返回The following statement has been executed successfully!資訊表示名為my_db的Paimon資料庫以及名為my_tbl的Paimon表建立成功。

步驟三：向Paimon表寫入資料

在資料開發 > ETL頁面的作業草稿頁簽，建立空白流作業草稿，詳情請參見SQL作業開發。將如下插入語句複製到SQL編輯器。

-- Paimon結果表在每次檢查點完成之後才會正式提交資料。
-- 此處將檢查點間隔縮短為10s，是為了更快地提交資料。
-- 在生產環境下，系統檢查點的間隔與兩次系統檢查點之間的最短時間間隔根據業務對延時要求的不同，一般設定為1分鐘到10分鐘。
SET 'execution.checkpointing.interval'='10s';
INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108',1,'apple'), ('20240108',2,'banana'), ('20240109',1,'cat'), ('20240109',2,'dog');

在SQL編輯地區右上方，單擊部署，在部署新版本對話方塊中，根據需要填寫或選中相關內容，單擊確定。
在營運中心 > 作業營運頁面，單擊目標作業名稱操作列下的啟動，選擇無狀態啟動，單擊啟動。
當作業狀態變為已完成，表示已完成資料的寫入。

步驟四：從Paimon表中流式消費資料

建立空白流作業草稿，將如下SQL代碼複製到SQL編輯器，使用print連接器，輸出my_tbl表中所有資料到日誌中。

CREATE TEMPORARY TABLE Print (
  dt STRING,
  id BIGINT,
  content STRING
) WITH (
  'connector' = 'print'
);
INSERT INTO Print SELECT * FROM `my-catalog`.`my_db`.`my_tbl`;

在SQL編輯地區右上方，單擊部署，在部署新版本對話方塊中，根據需要填寫或選中相關內容，單擊確定。
在營運中心 > 作業營運頁面，單擊目標作業操作列下的啟動，選擇無狀態啟動，單擊啟動。
在作業營運詳情頁面，查看Flink計算結果。
1. 在營運中心 > 作業營運頁面，單擊目標作業名稱。
2. 在作業日誌頁簽下的作業記錄頁面，單擊運行Task Managers頁簽下的Path, ID。
3. 單擊Stdout後，查看消費的Paimon資料。

步驟五：向Paimon表更新資料

建立空白流作業草稿，將如下SQL代碼複製到SQL編輯器。

SET 'execution.checkpointing.interval' = '10s';
INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108', 1, 'hello'), ('20240109', 2, 'world');

在SQL編輯地區右上方，單擊部署，在部署新版本對話方塊中，根據需要填寫或選中相關內容，單擊確定。
在營運中心 > 作業營運頁面，單擊目標作業操作列下的啟動，選擇無狀態啟動，單擊啟動。
當作業狀態變為已完成，表示已完成資料的寫入。
在步驟四作業的作業營運頁面的Stdout頁簽下，查看向Paimon表中更新的資料。

步驟六（可選）：停止流式消費作業並清理資源

測試完畢後，如果您需要停止流式消費作業並清理資源，可參考以下步驟：

在營運中心 > 作業營運頁面，單擊目標作業操作列的停止，停止作業運行。

在查詢指令碼頁簽的文本編輯地區，輸入如下代碼，刪除Paimon資料檔案以及Paimon Catalog。

DROP DATABASE `my-catalog`.`my_db` CASCADE; --將刪除儲存在OSS上的整個Paimon資料庫的資料檔案。
DROP CATALOG `my-catalog`; --將Paimon Catalog從Realtime Compute開發控制台中繼資料中刪除，但對OSS上的資料檔案沒有影響。

返回The following statement has been executed successfully!資訊表示Paimon資料檔案以及Paimon Catalog刪除成功。

Realtime Compute for Apache Flink：Paimon快速開始：準系統