使用DLF在Flink上構建Apache Iceberg湖倉表 - Realtime Compute for Apache Flink

配置Iceberg Catalog後，您可以通過Realtime ComputeFlink版直接存取阿里雲資料湖構建（Data Lake Formation）中的Iceberg表。本文為您介紹如何在Realtime Compute開發控制台建立、查看與刪除Iceberg Catalog，以及管理Iceberg資料庫和Iceberg表。

注意事項

僅Realtime Compute引擎VVR 11.1及以上版本支援建立和配置Iceberg Catalog和Iceberg 表。
僅支援對接 DLF Catalog。

建立Iceberg DLF Catalog

在DLF上建立 Catalog，詳情請參見快速使用DLF。
1. DLF Catalog需要和Flink工作空間位於同一地區，否則無法在後續流程中完成關聯。

在Realtime Compute開發控制台上，建立Iceberg Catalog。

說明

這裡僅用於建立與DLF Catalog的映射串連，建立或刪除Catalog僅影響映射關係，不會對DLF中的實際資料產生影響。
通過Iceberg REST 在DLF Catalog建立的表均為Iceberg表。

登入Realtime Compute控制台。
單擊目標工作空間操作列下的控制台，進入對應的工作空間。
在左側導覽列，單擊資料開發 > 資料查詢。

在資料查詢文本編輯地區，輸入以下SQL，單擊頁面右下角的執行環境，選擇對應版本的Session叢集（VVR 11.2.0及以上引擎版本），執行SQL，即可以Iceberg REST建立DLF Catalog。

CREATE CATALOG `catalog_name` 
 WITH (
    'type' = 'iceberg',
    'catalog-type' = 'rest',
    'uri' = 'http://cn-hangzhou-vpc.dlf.aliyuncs.com/iceberg',
    'warehouse' = 'iceberg_test',
    'rest.signing-region' = 'cn-hangzhou',
    'io-impl' = 'org.apache.iceberg.rest.DlfFileIO'
);

參數說明如下：

參數	描述	是否必填	樣本
type	類型，固定為iceberg。從自訂Jar自動解析，請勿更改。	是	iceberg
catalog-type	Catalog類型，固定為rest。	是	rest
token.provider	Token提供方，固定為dlf。	是	dlf
uri	訪問DLF Rest Catalog的URI，詳見Iceberg REST。	是	http://cn-hangzhou-vpc.dlf.aliyuncs.com/iceberg
warehouse	DLF Catalog名稱。	是	iceberg_test
rest.signing-region	DLF的Region ID，詳見服務存取點。	是	cn-hangzhou
io-impl	固定值：org.apache.iceberg.rest.DlfFileIO	是	org.apache.iceberg.rest.DlfFileIO

管理Iceberg資料庫

在資料查詢文本編輯地區輸入如下命令，選中代碼後單擊運行。

建立資料庫

建立Iceberg Catalog後，Catalog中會自動建立名為default的資料庫。

--my-catalog需要替換為您建立的Catalog名稱。
USE CATALOG `my-catalog`;

--my_db需要替換為自訂的database英文名稱。
CREATE DATABASE `my_db`;

刪除資料庫

重要

無法刪除DLF類型Catalog下的default資料庫

--my-catalog需要替換為您建立的Catalog名稱。
USE CATALOG `my-catalog`;

--my_db需要替換為您想要刪除的資料庫名稱。
DROP DATABASE `my_db`; --只需要刪除內部沒有任何錶的資料庫。
DROP DATABASE `my_db` CASCADE; --刪除資料庫且同時刪除資料庫下的所有表。

管理Iceberg表

建立表

說明

Iceberg Catalog配置完成後，您就可以在作業中引用Iceberg Catalog表資訊，作為源表、結果表和維表時，無需聲明表的DDL。在SQL命令方式中，您可以直接使用Iceberg Catalog表名稱的完整格式${Iceberg-catalog-name}.${Iceberg-db-name}.${Iceberg-table-name}，也可以用use catalog ${Iceberg-catalog-name}和use ${Iceberg-db-name}語句先聲明Catalog名稱和資料庫名稱，後面的SQL裡唯寫表名稱${Iceberg-table-name}即可。

通過CREATE TABLE語句建立表

在資料查詢文本編輯地區輸入如下命令後，選中代碼後單擊運行。

在名為my-catalog的Catalog下的my_db的資料庫中，建立一張分區表，其中分區鍵為dt，主鍵為dt、shop_id和user_id。程式碼範例如下。

--my-catalog需要替換為您建立的Iceberg Catalog名稱。
--my_db需要替換為您想要使用的database名稱。
--my_tbl也可以改為自訂的英文名稱。
CREATE TABLE `iceberg_test`.`test`.`my_tbl` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt)
;

修改表結構

在資料查詢文本編輯地區輸入如下命令，選中代碼後單擊運行。

操作	程式碼範例
臨時修改表參數	您可以通過在表名後添加SQL Hint的方式，在寫入表時臨時修改表參數。臨時修改的表參數僅對當前SQL作業生效。在寫入my_table表時，臨時將`upsert-enabled`設為true。 `INSERT INTO tableName /+ OPTIONS('upsert-enabled'='true') / SELECT ...;` 在消費my_table表時，臨時將`monitor-interval`設為10s。 `SELECT * FROM my_table /+ OPTIONS('monitor-interval'='10s') /`
重新命名表	將表的名稱從my_table修改為my_table_new。 `ALTER TABLE my_table RENAME TO my_table_new;`
添加新列	將類型為INT的c1列和類型為STRING的c2列添加在my_table表的末尾。 `ALTER TABLE my_table ADD (c1 INT, c2 STRING);`
重新命名列	將my_table表中名稱為c0的列重新命名為c1。 `ALTER TABLE my_table RENAME c0 TO c1;`
刪除列	將my_table表中c1和c2兩列刪除。 `ALTER TABLE my_table DROP (c1, c2);`
修改列注釋	將my_table表中buy_count列的注釋改為this is buy count。 `ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'this is buy count';`
修改列順序	將my_table表中名稱為col_a，類型為DOUBLE的列移動到開頭。 `ALTER TABLE my_table MODIFY col_a DOUBLE FIRST;` 將my_table表中名稱為col_a，類型為DOUBLE的列移動到col_b列的後面。 `ALTER TABLE my_table MODIFY col_a DOUBLE AFTER col_b;`

刪除表

在資料查詢文本編輯地區輸入如下命令，選中代碼後單擊運行。

--my-catalog需要替換為您建立的Iceberg Catalog名稱。
--my_db需要替換為您想要使用的database名稱。
--my_tbl需要替換為您建立的Iceberg Catalog表名稱。
DROP TABLE `my-catalog`.`my_db`.`my_tbl`;

出現The following statement has been executed successfully!的提示資訊說明Iceberg表刪除成功。

、

查看或刪除Iceberg Catalog

在Realtime Compute控制台，單擊目標工作空間操作列的控制台。
在資料管理頁面，查看或刪除Iceberg Catalog。
- 查看：在Catalog列表頁面，查看Catalog名稱和類型。如果您需要查看Catalog下的資料庫和表，請單擊查看。
- 刪除：在Catalog列表頁面，單擊目標Catalog名稱對應操作列的刪除。
  說明
  刪除Iceberg Catalog只會刪除Flink專案空間資料管理中的記錄，不會影響Iceberg表的資料檔案。刪除Catalog後，只需要重新執行建立Iceberg Catalog，即可重新使用Catalog中的Iceberg表。
  您也可以在資料查詢文本編輯地區輸入DROP CATALOG <catalog name>;後，選中代碼後單擊運行。