Iceberg 表引擎 - ClickHouse Documentation

我们建议在 ClickHouse 中处理 Iceberg 数据时使用 Iceberg 表函数。Iceberg 表函数目前已具备足够的功能，可为 Iceberg 表提供部分只读接口。Iceberg 表引擎已可用，但可能存在一些限制。ClickHouse 最初并不是为支持 schema 会被外部更改的表而设计的，这可能会影响 Iceberg 表引擎的功能。因此，一些适用于常规表的功能可能无法使用，或者无法正常工作，尤其是在使用旧版 analyzer 时。为获得最佳兼容性，建议使用 Iceberg 表函数；同时，我们也会继续改进对 Iceberg 表引擎的支持。

该引擎为 Amazon S3、Azure、HDFS 以及本地存储中的现有 Apache Iceberg 表提供只读集成。

创建表

请注意，Iceberg 表必须已存在于存储中；此命令不接受用于创建新表的 DDL 参数。

CREATE TABLE iceberg_table_s3
    ENGINE = IcebergS3(url,  [, NOSIGN | access_key_id, secret_access_key, [session_token]], format, [,compression], [,extra_credentials])

CREATE TABLE iceberg_table_azure
    ENGINE = IcebergAzure(connection_string|storage_account_url, container_name, blobpath, [account_name, account_key, format, compression])

CREATE TABLE iceberg_table_hdfs
    ENGINE = IcebergHDFS(path_to_table, [,format] [,compression_method])

CREATE TABLE iceberg_table_local
    ENGINE = IcebergLocal(path_to_table, [,format] [,compression_method])

引擎参数

这些参数的说明分别与 S3、AzureBlobStorage、HDFS 和 File 表引擎中的参数说明一致。 format 表示 Iceberg 表中数据文件的格式。对于 IcebergS3，可使用可选参数 extra_credentials 传递 role_arn，以便在 ClickHouse Cloud 中进行基于角色的访问。有关配置步骤，请参阅安全访问 S3。可以使用命名集合指定引擎参数。

示例

CREATE TABLE iceberg_table ENGINE=IcebergS3('http://test.s3.amazonaws.com/clickhouse-bucket/test_table', 'test', 'test')

使用命名集合：

<clickhouse>
    <named_collections>
        <iceberg_conf>
            <url>http://test.s3.amazonaws.com/clickhouse-bucket/</url>
            <access_key_id>test</access_key_id>
            <secret_access_key>test</secret_access_key>
        </iceberg_conf>
    </named_collections>
</clickhouse>

CREATE TABLE iceberg_table ENGINE=IcebergS3(iceberg_conf, filename = 'test_table')

别名

Iceberg 表引擎会根据 disk 设置自动检测存储后端，并相应路由到 IcebergS3、IcebergAzure 或 IcebergLocal。如果未指定 disk，则默认使用 IcebergS3 实现。

数据类型

下表展示了在进行 schema 推断时 (用于读取) Iceberg 数据类型与 ClickHouse 数据类型之间的映射关系。

基本类型

Iceberg 类型	ClickHouse 类型	说明
`boolean`	`Bool`
`int`	`Int32`
`long`, `bigint`	`Int64`
`float`	`Float32`
`double`	`Float64`
`date`	`Date32`
`time`	`Int64`	自午夜起的微秒数
`timestamp`	`DateTime64(6)`	微秒，无时区
`timestamptz`	`DateTime64(6, 'UTC')`	微秒，UTC 时区
`timestamp_ns`	`DateTime64(9)`	纳秒，无时区 (仅适用于 Iceberg v3 及以上版本)
`timestamptz_ns`	`DateTime64(9, 'UTC')`	纳秒，UTC 时区 (仅适用于 Iceberg v3 及以上版本)
`string`, `binary`	`String`
`uuid`	`UUID`
`fixed(N)`	`FixedString(N)`
`decimal(P, S)`	`Decimal(P, S)`

复合类型

Iceberg 类型	ClickHouse 类型
`list`	`Array`
`map`	`Map`
`struct`	`Tuple`

Schema 演进

ClickHouse 支持读取 schema 会随时间演进的 Iceberg 表。这包括列被添加、删除或重新排序的表，以及列从必填变为可为空的表。此外，还支持以下类型转换：

int -> long
float -> double
decimal(P, S) -> decimal(P’, S) where P’ > P.

目前，尚不支持更改嵌套结构，也不支持更改数组和 Map 中元素的类型。要读取使用动态 schema 推断创建后 schema 又发生变化的表，请在创建表时设置 allow_dynamic_metadata_for_data_lakes = true。

分区裁剪

ClickHouse 支持在针对 Iceberg 表的 SELECT 查询中进行分区裁剪，这有助于通过跳过无关的数据文件来优化查询性能。要启用分区裁剪，请设置 use_iceberg_partition_pruning = 1。有关 Iceberg 分区裁剪的更多信息，请参阅 https://iceberg.apache.org/spec/#partitioning

时间旅行

ClickHouse 支持 Iceberg 表的时间旅行，让你能够通过特定的时间戳或快照 ID 查询历史数据。

包含已删除行的表的处理

ClickHouse 支持读取使用以下删除方法的 Iceberg 表：

位置删除
等值删除 (自 25.8+ 版本起支持)

以下删除方法不支持：

删除向量 (在 v3 中引入)

基本用法

 SELECT * FROM example_table ORDER BY 1 
 SETTINGS iceberg_timestamp_ms = 1714636800000

 SELECT * FROM example_table ORDER BY 1 
 SETTINGS iceberg_snapshot_id = 3547395809148285433

注意：在同一个查询中，不能同时指定 iceberg_timestamp_ms 和 iceberg_snapshot_id 参数。

重要注意事项

快照通常会在以下情况下创建：
- 向表中写入新数据时
- 执行某种数据合并整理时
schema 变更通常不会创建快照——这会在对经历过 schema 演进的表使用时间旅行时带来一些重要特性。

示例场景

所有场景均使用 Spark 编写，因为 CH 目前尚不支持写入 Iceberg 表。

场景 1：没有新快照的 schema 变更

考虑以下操作顺序：

 -- 创建一个包含两列的表
  CREATE TABLE IF NOT EXISTS spark_catalog.db.time_travel_example (
  order_number int, 
  product_code string
  ) 
  USING iceberg 
  OPTIONS ('format-version'='2')

-- 向表中插入数据
  INSERT INTO spark_catalog.db.time_travel_example VALUES 
    (1, 'Mars')

  ts1 = now() // 一段伪代码

-- 修改表以添加新列
  ALTER TABLE spark_catalog.db.time_travel_example ADD COLUMN (price double)
 
  ts2 = now()

-- 向表中插入数据
  INSERT INTO spark_catalog.db.time_travel_example VALUES (2, 'Venus', 100)

   ts3 = now()

-- 查询各时间戳对应的表数据
  SELECT * FROM spark_catalog.db.time_travel_example TIMESTAMP AS OF ts1;

+------------+------------+
|order_number|product_code|
+------------+------------+
|           1|        Mars|
+------------+------------+
  SELECT * FROM spark_catalog.db.time_travel_example TIMESTAMP AS OF ts2;

+------------+------------+
|order_number|product_code|
+------------+------------+
|           1|        Mars|
+------------+------------+

  SELECT * FROM spark_catalog.db.time_travel_example TIMESTAMP AS OF ts3;

+------------+------------+-----+
|order_number|product_code|price|
+------------+------------+-----+
|           1|        Mars| NULL|
|           2|       Venus|100.0|
+------------+------------+-----+

不同时间戳下的查询结果：

在 ts1 和 ts2 时：仅显示最初的两列
在 ts3 时：显示全部三列，其中第一行的 price 为 NULL

场景 2：历史 schema 与当前 schema 的差异

在当前时刻执行时间旅行查询时，显示出的 schema 可能与当前表的 schema 不同：

-- 创建表
  CREATE TABLE IF NOT EXISTS spark_catalog.db.time_travel_example_2 (
  order_number int, 
  product_code string
  ) 
  USING iceberg 
  OPTIONS ('format-version'='2')

-- 向表中插入初始数据
  INSERT INTO spark_catalog.db.time_travel_example_2 VALUES (2, 'Venus');

-- 修改表以添加新列
  ALTER TABLE spark_catalog.db.time_travel_example_2 ADD COLUMN (price double);

  ts = now();

-- 使用时间戳语法查询当前时刻的表数据

  SELECT * FROM spark_catalog.db.time_travel_example_2 TIMESTAMP AS OF ts;

    +------------+------------+
    |order_number|product_code|
    +------------+------------+
    |           2|       Venus|
    +------------+------------+

-- 查询当前时刻的表数据
  SELECT * FROM spark_catalog.db.time_travel_example_2;
    +------------+------------+-----+
    |order_number|product_code|price|
    +------------+------------+-----+
    |           2|       Venus| NULL|
    +------------+------------+-----+

之所以会出现这种情况，是因为 ALTER TABLE 不会创建新的快照；对于当前表，Spark 获取 schema_id 的值时，取自最新的元数据文件，而不是某个快照。

场景 3：历史 schema 与当前 schema 的差异

第二点是，进行时间旅行时，你无法获取该表在尚未写入任何数据之前的状态：

-- 创建一张表
  CREATE TABLE IF NOT EXISTS spark_catalog.db.time_travel_example_3 (
  order_number int, 
  product_code string
  ) 
  USING iceberg 
  OPTIONS ('format-version'='2');

  ts = now();

-- 查询表在特定时间戳时的状态
  SELECT * FROM spark_catalog.db.time_travel_example_3 TIMESTAMP AS OF ts; -- 报错：Cannot find a 快照 older than ts.

在 ClickHouse 中，其行为与 Spark 保持一致。你可以直接把 Spark 的 Select 查询理解为 ClickHouse 的 Select 查询，两者的工作方式是一样的。

元数据文件解析

在 ClickHouse 中使用 Iceberg 表引擎时，系统需要找到描述 Iceberg 表结构的正确 metadata.json 文件。下面介绍这一解析过程的工作原理：

候选文件搜索

直接指定路径：

如果设置了 iceberg_metadata_file_path，系统会将其与 Iceberg 表目录路径拼接，使用这个精确路径。
提供此设置后，其他所有解析设置都会被忽略。

表 UUID 匹配：

如果指定了 iceberg_metadata_table_uuid，系统将：
- 只检查 metadata 目录中的 .metadata.json 文件
- 筛选出包含 table-uuid 字段且与指定 UUID 匹配 (不区分大小写) 的文件

默认搜索：

如果上述两个设置都未提供，则 metadata 目录中的所有 .metadata.json 文件都会作为候选文件

选择最新的文件

根据上述规则识别出候选文件后，系统会进一步确定其中哪个文件最新：

如果启用了 iceberg_recent_metadata_file_by_last_updated_ms_field：
- 选择 last-updated-ms 值最大的文件
否则：
- 选择版本号最大的文件
- (在格式为 V.metadata.json 或 V-uuid.metadata.json 的文件名中，版本号显示为 V)

注意：上述提到的所有设置 (除非另有明确说明) 均为引擎级设置，必须在创建表时按如下所示指定：

CREATE TABLE example_table ENGINE = Iceberg(
    's3://bucket/path/to/iceberg_table'
) SETTINGS iceberg_metadata_table_uuid = '6f6f6407-c6a5-465f-a808-ea8900e35a38';

注意：虽然 Iceberg 目录通常负责元数据解析，但 ClickHouse 的 Iceberg 表引擎会直接将存储在 S3 中的文件识别为 Iceberg 表，因此理解这些解析规则非常重要。

数据缓存

Iceberg 表引擎和表函数支持数据缓存，与 S3、AzureBlobStorage 和 HDFS 存储相同。请参见此处。

元数据缓存

Iceberg 表引擎和表函数支持元数据缓存，可缓存 manifest 文件、manifest 列表和 metadata json 的相关信息。缓存存储在内存中。此功能由设置 use_iceberg_metadata_files_cache 控制，默认启用。

异步元数据预取

在创建 Iceberg 表时，可通过设置 iceberg_metadata_async_prefetch_period_ms 启用异步元数据预取。如果将其设为 0 (默认值) ，或未启用元数据缓存，则会禁用异步预取。要启用此功能，需要指定一个非零的毫秒值，表示两次预取周期之间的时间间隔。启用后，服务器会在后台周期性执行一项操作，列出远程目录并检测新的元数据版本。随后会对其进行解析，并递归遍历快照，拉取活动的 manifest 列表和 manifest 文件。已存在于元数据缓存中的文件不会被重复下载。每个预取周期结束时，最新的元数据快照都会存放在元数据缓存中。

CREATE TABLE example_table ENGINE = Iceberg(
    's3://bucket/path/to/iceberg_table'
) SETTINGS
    iceberg_metadata_async_prefetch_period_ms = 60000;

为了在读取操作中尽可能利用异步元数据预取，应将 iceberg_metadata_staleness_ms 指定为查询参数或会话参数。默认情况下 (0，即未指定) ，在每个查询的上下文中，服务器都会从远程目录拉取最新元数据。通过指定可容忍的元数据过期程度，服务器便可在不调用远程目录的情况下使用已缓存的元数据快照版本。如果缓存中存在元数据版本，且其下载时间处于给定的过期窗口内，则会使用该版本来处理查询。否则，将从远程目录拉取最新版本。

SELECT count() FROM icebench_table WHERE ...
SETTINGS iceberg_metadata_staleness_ms=120000

注意：异步元数据预取在 ICEBERG_SCEDULE_POOL 中运行，这是一个服务器端线程池，用于对活动 Iceberg 表执行后台操作。该线程池的大小由服务器配置参数 iceberg_background_schedule_pool_size 控制 (默认值为 10) 。注意：目前预期是，如果启用了异步预取，元数据缓存的大小应足以完整容纳所有活动表的最新元数据快照。

另请参见

Iceberg 表函数

​创建表

​引擎参数

​示例

​别名

​数据类型

​基本类型

​复合类型

​Schema 演进

​分区裁剪

​时间旅行

​包含已删除行的表的处理

​基本用法

​重要注意事项

​示例场景

​场景 1：没有新快照的 schema 变更

​场景 2：历史 schema 与当前 schema 的差异

​场景 3：历史 schema 与当前 schema 的差异

​元数据文件解析

​候选文件搜索

​选择最新的文件

​数据缓存

​元数据缓存

​异步元数据预取

​另请参见

创建表