通過Spark SQL讀寫Iceberg外表

如何選擇Catalog模式

• 僅在AnalyticDB for MySQL內部使用：三種模式均可，推薦使用湖儲存模式（若已開通湖儲存）或預設的MDS Catalog模式，配置最簡單。

• 需要跨產品協同讀寫同一張表（例如EMR Spark寫入後由AnalyticDB for MySQL查詢，或Flink寫入後由Spark查詢）：建議使用Hadoop Catalog或湖儲存模式。不同Catalog模式的中繼資料相互獨立，跨模式混合寫入同一張表會導致中繼資料不一致，詳情請參見注意事項。

表名引用方式

AnalyticDB for MySQL支援以下兩種方式引用Iceberg表：

• 兩層結構（推薦）：資料庫名.表名，例如my_db.my_table。系統會自動識別表類型並路由到對應的Iceberg Catalog，無需指定Catalog名稱。

• 三層結構：Catalog名.資料庫名.表名，例如iceberg.my_db.my_table。適用於需要明確指定Catalog的情境。

Spark 3.5及以上版本中，系統自動註冊了以下Iceberg Catalog名稱：

樣本（以下三種寫法等價）

SELECT * FROM my_db.my_table;
SELECT * FROM iceberg.my_db.my_table;
SELECT * FROM adb_lakehouse_prod.my_db.my_table;

步驟一：進入資料開發

1. 登入雲原生資料倉儲AnalyticDB MySQL控制台，在左上方選擇叢集所在地區。在左側導覽列，單擊集群清單，然後單擊目的地組群ID。

2. 配置Spark參數。

   根據所選的Catalog模式，需要配置不同模式的SET參數：

   Catalog模式	SET參數
   MDS Catalog（預設）	SET spark.adb.version=3.5;
   Hadoop Catalog	SET spark.adb.version=3.5; SET spark.iceberg.warehouse=oss://testBucketName/iceberg/;
   湖儲存	SET spark.adb.lakehouse.enabled=true;

   SET參數的配置方式因資源群組類型而異：

   • Job型資源群組 ：直接將SET語句寫在SQL語句前面即可。如本文後續SQL樣本所示。

   • Interactive型資源群組 ：需要將Spark參數預先配置到資源群組中。配置完成後，SQL語句中無需再添加對應的SET語句。

     1. 在左側導覽列單擊集群管理 > 資源管理。

     2. 在目標資源群組所在行操作列，單擊修改。

     3. 在修改資源組面板中更新Spark 配置，然後單擊確定。

        當資源群組狀態變為運行中時，修改生效。

3. 在左側導覽列，單擊作業開發 > SQL開發。

4. 在SQLConsole視窗，選擇Spark引擎和資源群組（Job型資源群組或Spark引擎的Interactive型資源群組）。

步驟二：建立外庫與Iceberg外表

請根據所選的Catalog模式，按照對應方式建立外庫和Iceberg外表。

步驟三：寫入或刪除Iceberg外表資料

以下樣本以MDS Catalog模式為主進行說明。

DELETE FROM adb_iceberg_db.test_iceberg_tbl
WHERE id = 1;

DELETE FROM adb_iceberg_db.test_iceberg_tbl
WHERE age = 20;

步驟四：查詢Iceberg外表資料

SET spark.adb.version=3.5;

SELECT * FROM adb_iceberg_db.test_iceberg_tbl;

返回樣本

+---+----+---+
|id |name|age|
+---+----+---+
|2  |box|30 |
+---+----+---+

步驟五：刪除Iceberg外表

SET spark.adb.version=3.5;

-- 僅刪除表的中繼資料，不刪除底層資料檔案
DROP TABLE adb_iceberg_db.test_iceberg_tbl;

-- 同時刪除表的中繼資料和底層資料檔案
DROP TABLE adb_iceberg_db.test_iceberg_tbl PURGE;

SET spark.adb.version=3.5;

DROP TABLE adb_iceberg_db.test_iceberg_tbl;

SET spark.adb.lakehouse.enabled=true;

DROP TABLE adb_external_db_iceberg.test_iceberg_tbl;

查詢中繼資料表

Iceberg為每張表自動維護了多種中繼資料表，您可以通過表名.中繼資料表名的方式進行查詢。

SET spark.adb.version=3.5;

-- 查看錶的快照資訊（快照ID、時間戳記、操作類型等）
SELECT * FROM adb_iceberg_db.test_iceberg_tbl.snapshots;

-- 查看錶的變更歷史
SELECT * FROM adb_iceberg_db.test_iceberg_tbl.history;

-- 查看當前快照中的資料檔案資訊
SELECT * FROM adb_iceberg_db.test_iceberg_tbl.files;

-- 查看manifest條目（每個檔案的詳細引用資訊）
SELECT * FROM adb_iceberg_db.test_iceberg_tbl.entries;

-- 查看分區統計資訊（記錄數、檔案數等）
SELECT * FROM adb_iceberg_db.test_iceberg_tbl.partitions;

常用的中繼資料表包括snapshots、history、files、manifests、entries、partitions和refs等。此外，以all_為首碼的中繼資料表（如all_data_files、all_manifests、all_entries）可查看包含歷史快照在內的全量資訊。完整列表請參見Apache Iceberg官方文檔。

使用預存程序（Procedure）

您可以通過CALL語句調用預存程序進行表維護。

-- 兩層結構：直接使用system命名空間
CALL system.procedure_name('資料庫名.表名', 其他參數...);

-- 三層結構：需加上Catalog名首碼
CALL iceberg.system.procedure_name('資料庫名.表名', 其他參數...);

以下是常用的表維護操作樣本。

• 合并小檔案（rewrite_data_files）

  SET spark.adb.version=3.5;
  
  -- 使用預設binpack策略合并小檔案
  CALL system.rewrite_data_files('adb_iceberg_db.test_iceberg_tbl');
  
  -- 使用sort策略並指定排序欄位，僅對指定分區生效
  CALL system.rewrite_data_files(
    table => 'adb_iceberg_db.test_iceberg_tbl',
    strategy => 'sort',
    sort_order => 'id ASC NULLS FIRST',
    where => 'age = 10'
  );

• 最佳化manifest檔案（rewrite_manifests）

  CALL system.rewrite_manifests('adb_iceberg_db.test_iceberg_tbl');

• 到期快照清理（expire_snapshots）

  CALL system.expire_snapshots(
    table => 'adb_iceberg_db.test_iceberg_tbl',
    older_than => TIMESTAMP '2024-01-01 00:00:00',
    retain_last => 5
  );

• 快照復原（rollback_to_snapshot）

  -- 先查詢快照列表擷取目標snapshot_id
  SELECT * FROM adb_iceberg_db.test_iceberg_tbl.snapshots;
  
  -- 復原到指定快照
  CALL system.rollback_to_snapshot('adb_iceberg_db.test_iceberg_tbl', 1234567890);

除上述預存程序外，還支援remove_orphan_files（清理孤立檔案）和rollback_to_timestamp（按時間復原）等操作。完整列表請參見Apache Iceberg官方文檔。

AnalyticDB for MySQL增強預存程序

AnalyticDB for MySQL在標準Iceberg預存程序基礎上，額外提供了資料歸檔和生命週期管理能力。

• 資料歸檔（archive_files）

  將冷資料歸檔到低成本儲存類型，降低儲存費用。

  -- 歸檔30天前的資料到Archive儲存（dry_run=true表示僅預覽影響範圍，不實際執行）
  CALL system.archive_files(
    table => 'adb_iceberg_db.test_iceberg_tbl',
    storage_type => 'Archive',
    start_days => 30,
    dry_run => true
  );

• 檢查歸檔狀態（check_archive_status）

  查看各儲存類型下的檔案分布情況。

  CALL system.check_archive_status(
    table => 'adb_iceberg_db.test_iceberg_tbl',
    group_by_partition => true
  );

• 資料解凍（restore_files）

  將已歸檔的資料恢複到標準儲存類型。

  CALL system.restore_files(
    table => 'adb_iceberg_db.test_iceberg_tbl',
    where => 'age = 10'
  );

• 分區生命週期管理（ttl）

  基於分區的最後更新時間，識別或刪除到期分區。

  -- 查看超過90天未更新的分區（dry_run=true表示僅預覽，不執行刪除）
  CALL system.ttl(
    table => 'adb_iceberg_db.test_iceberg_tbl',
    expire_after_days => 90,
    dry_run => true
  );

不同Catalog模式的中繼資料一致性

三種Catalog模式各自獨立管理Iceberg表的中繼資料。在AnalyticDB for MySQL內部，MDS Catalog能夠感知通過Hadoop Catalog或湖儲存模式建立的表，因此在不添加額外配置的情況下也可以讀取這些表。但這不意味著可以跨模式混合寫入。

各Catalog模式之間的跨模式讀寫行為：

運行環境配置

• 使用Iceberg相關功能時，Spark版本需為3.5及以上。如果叢集預設Spark版本不是3.5，請在SQL語句開頭添加SET spark.adb.version=3.5;。

• 根據所選的Catalog模式，需要配置不同模式的SET參數，不同資源群組的配置方式，請參見SET參數配置。

批處理與互動式模式

Spark SQL支援批處理和互動式兩種運行模式。請確認所使用的資源群組類型與運行模式相匹配。