欢迎光临
我们一直在努力
        找回密码
当前位置：171主机测评 > 服务器技术 > 正文

Apache Iceberg + Flink 快速上手实战：基于 Docker Compose 的 REST Catalog 环境搭建与 SQL 读写全流程

2026-10-26 分类：服务器技术 阅读(10) 评论(0)
  • 数据湖
  • 大数据
  • 数据存储

【免费下载链接】iceberg

Apache Iceberg

项目地址：
https://gitcode.com/gh_mirrors/icebe/iceberg

点击查看 免费下载

Apache Iceberg 是面向大数据分析场景的开源表格格式，而 Apache Flink 是主流的有状态流处理引擎，两者的结合让流批一体的数据湖分析成为现实。本文以官方提供的 Iceberg Flink Quickstart 环境为骨架，带你用一条 docker compose 命令在本机拉起 Flink 集群、Iceberg REST Catalog 与 MinIO 对象存储，并完整走通「建 Catalog → 建库建表 → 写入 → 查询 → 内联配置建表」的 SQL 实战链路。读完本文，你将掌握 Iceberg 在 Flink SQL Client 下的最小可用部署方式、核心配置参数的含义，以及 Iceberg 快照与元数据表的基本用法。

Quickstart 环境概览

官方 Quickstart 提供了一套开箱即用的 Docker Compose 编排（定义在 docker/iceberg-flink-quickstart/docker-compose.yml），整套环境包含三个核心部件：

组件作用
本地 Flink 集群（JobManager + TaskManager） 执行 Flink SQL，提供 sql-client.sh 交互式会话
Iceberg REST Catalog 管理 Iceberg 表的元数据，提供标准 REST 协议接口
MinIO（本地 S3 存储） 充当 S3 兼容对象存储，存放 Iceberg 的数据文件与元数据文件

Iceberg Flink Quickstart 容器架构概览

这套环境把「计算层（Flink）+ 元数据层（REST Catalog）+ 存储层（MinIO）」全部编排在同一 Docker Compose 网络 iceberg_net 中，组件之间通过容器名直接通信，无需任何外部依赖，是本机体验 Iceberg 的最快路径。

使用 Quickstart 之前，你需要先安装 Docker CLI（含 docker compose 插件）。

一键启动 Quickstart 环境

克隆 Iceberg 仓库并启动所有容器：

git clone https://github.com/apache/iceberg.git
cd iceberg
docker compose -f docker/iceberg-flink-quickstart/docker-compose.yml up -d –build

启动完成后，进入 Flink SQL Client 交互会话：

docker exec -it jobmanager ./bin/sql-client.sh

容器编排的源码级解读

从 docker-compose.yml 可以看到这套环境的启动顺序与关键配置：

  • jobmanager / taskmanager：基于自定义镜像构建，通过 FLINK_PROPERTIES 设置了 jobmanager.rpc.address: jobmanager、taskmanager.numberOfTaskSlots: 2 与 parallelism.default: 2；同时注入了 AWS_REGION、AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY 和 S3_ENDPOINT 环境变量，供 S3 FileIO 使用。JobManager 依赖 iceberg-rest 服务健康检查通过后才启动。
  • iceberg-rest：使用 apache/iceberg-rest-fixture 镜像（其构建脚本见 docker/iceberg-rest-fixture/Dockerfile），通过 CATALOG_WAREHOUSE=s3://warehouse/、CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO、CATALOG_S3_ENDPOINT 与 CATALOG_S3_ACCESS__KEY__ID 等环境变量完成自身配置，健康检查探测 http://localhost:8181/v1/config。
  • minio：quay.io/minio/minio 镜像，根用户/密码为 admin / password，监听 9000 端口（对象存储 API）与 9001 端口（管理控制台）。
  • create-bucket：一次性服务，等待 MinIO 就绪后调用 mc 创建并公开 warehouse 桶，为 REST Catalog 的 s3://warehouse/ 仓库路径提供实际存储位置。

镜像内置的 Iceberg 依赖

Flink SQL Client 之所以能直接识别 'type' = 'iceberg'，是因为基础镜像（见 docker/iceberg-flink-quickstart/Dockerfile）在 apache/flink:2.0-java21 基础上预装了：

  • Iceberg Flink runtime（iceberg-flink-runtime-2.0-1.10.1.jar）：Iceberg 的 Flink 集成核心 jar，包含 SQL Catalog、Source/Sink 与类型转换等能力；
  • Iceberg AWS bundle（iceberg-aws-bundle-1.10.1.jar）：提供 S3 FileIO 等 AWS 相关实现；
  • Hadoop client API 与 runtime bundle（3.4.2）：提供 Hadoop 兼容层依赖。

构建参数 FLINK_VERSION、ICEBERG_FLINK_RUNTIME_VERSION、ICEBERG_VERSION、HADOOP_VERSION 均可覆盖，默认值分别为 2.0、2.0、1.10.1、3.4.2。如果你在仓库本地自行构建 Flink runtime jar，iceberg-flink-runtime 模块位于 flink/v2.1/flink-runtime、flink/v2.2/flink-runtime、flink/v2.3/flink-runtime 等目录（见 flink 模块结构）。

在 Flink 中创建 Iceberg Catalog

Iceberg 支持多种 Catalog 后端来跟踪表，例如 JDBC、Hive Metastore、AWS Glue 等。本指南使用 REST Catalog + S3 存储 的组合：REST Catalog 负责元数据，MinIO 负责数据文件。Catalog 相关细节可进一步查阅 Flink 配置文档。

首先在 Flink SQL Client 中定义一个 Flink Catalog，该 Catalog 下的表都将存储在 S3 对象存储中：

CREATE CATALOG iceberg_catalog WITH (
'type' = 'iceberg',
'catalog-impl' = 'org.apache.iceberg.rest.RESTCatalog',
'uri' = 'http://iceberg-rest:8181',
'warehouse' = 's3://warehouse/',
'io-impl' = 'org.apache.iceberg.aws.s3.S3FileIO',
's3.endpoint' = 'http://minio:9000',
's3.access-key-id' = 'admin',
's3.secret-access-key' = 'password',
's3.path-style-access' = 'true'
);

Catalog 参数含义

参数说明
type 必须为 iceberg，声明这是 Iceberg 类型的 Flink Catalog
catalog-impl 自定义 Catalog 实现的全限定类名，此处指向 REST Catalog 实现 org.apache.iceberg.rest.RESTCatalog
uri REST Catalog 服务的 URL。Quickstart 中通过容器名 iceberg-rest 访问 8181 端口
warehouse 仓库根路径，Iceberg 表的数据文件与元数据文件都将存储于此
io-impl 文件 IO 实现类，此处为 org.apache.iceberg.aws.s3.S3FileIO，让 Iceberg 通过 S3 协议访问存储
s3.endpoint S3 兼容服务地址，指向 MinIO（http://minio:9000）
s3.access-key-id / s3.secret-access-key MinIO 的访问凭证（admin / password）
s3.path-style-access 是否使用 path-style 寻址（true），MinIO 等本地 S3 兼容服务通常需要开启

此外，Flink 侧还有一批可选的全局 Catalog 属性（详见 Flink 配置文档）：

属性可选值说明
catalog-type hive、hadoop、rest 等 使用内置 Catalog 实现时指定；使用 catalog-impl 自定义实现时须留空
property-version 1 属性格式版本号，用于向后兼容
cache-enabled true / false 是否启用 Catalog 缓存，默认 true
cache.expiration-interval-ms 毫秒数 Catalog 条目的本地缓存过期时间，-1 表示永不过期（默认），不允许设为 0

在 Catalog 中创建一个数据库：

CREATE DATABASE IF NOT EXISTS iceberg_catalog.nyc;

数据库创建后，后续建表即可使用 iceberg_catalog.nyc.<表名> 的三段式命名。

创建 Iceberg 表

使用 Flink 的 CREATE TABLE 命令创建第一张 Iceberg 表（完整 DDL 说明见 Flink DDL 文档）。下面创建 iceberg_catalog.nyc.taxis，其中 iceberg_catalog 是 Catalog 名，nyc 是数据库名，taxis 是表名：

CREATE TABLE iceberg_catalog.nyc.taxis
(
vendor_id BIGINT,
trip_id BIGINT,
trip_distance FLOAT,
fare_amount DOUBLE,
store_and_fwd_flag STRING
);

Iceberg Catalog 支持 Flink SQL DDL 的完整能力，包括：

  • CREATE TABLE … PARTITIONED BY：按列分区（注意 Flink 暂不支持 Iceberg 的隐藏分区，不能对列做函数表达式分区）；
  • ALTER TABLE：支持 SET / RESET 表属性、ADD / DROP / RENAME / MODIFY 列，以及 RENAME TO 重命名表；修改分区字段、水印与计算列暂不支持；
  • DROP TABLE：删除表。

创建表时还可以在 WITH (…) 中指定表配置（例如 'format-version'='2'、'location'='…'），这些配置会写入 Iceberg 表属性并持久化。

向表写入数据

表创建完成后即可插入记录。这里有一个关键前提：Flink 依赖 checkpoint 保证数据持久性与精确一次（exactly-once）语义。Iceberg 的提交（commit）发生在 Flink checkpoint 成功之后，如果没有开启 checkpointing，Iceberg 的数据和元数据可能无法完整提交到存储中。

在 SQL Client 中开启周期性 checkpoint：

SET 'execution.checkpointing.interval' = '10s';

然后写入几条数据：

INSERT INTO iceberg_catalog.nyc.taxis
VALUES (1, 1000371, 1.8, 15.32, 'N'), (2, 1000372, 2.5, 22.15, 'N'), (2, 1000373, 0.9, 9.01, 'N'), (1, 1000374, 8.4, 42.13, 'Y');

从实现层面看（见 Flink 写入文档），Iceberg 的 Flink Sink 保证 exactly-once：写入器在 checkpoint 时刷新并上传数据文件，提交器（committer operator）在 notifyCheckpointComplete 回调中完成 Iceberg 提交。因此 checkpoint 间隔在流式写入场景下也决定了 Iceberg 快照的提交频率。除了 INSERT INTO 追加写入，Flink 还支持批作业下的 INSERT OVERWRITE 原子覆盖写，以及基于主键的 UPSERT（v2 表格式）。

从表读取数据

读取一张表，直接使用 Iceberg 表的完整名称：

SELECT * FROM iceberg_catalog.nyc.taxis;

除了普通查询，还可以借助 Iceberg 的元数据表（metadata tables） 检查表的状态。Quickstart 自带的验证脚本 test.sql 展示了几个常用元数据查询：

— 查看表的全部快照
SELECT * FROM iceberg_catalog.nyc.`taxis$snapshots`;

— 查看数据文件及其格式与记录数
SELECT content, file_path, file_format, record_count
FROM iceberg_catalog.nyc.`taxis$files`;

— 查看表的修改历史
SELECT * FROM iceberg_catalog.nyc.`taxis$history`;

Iceberg 的每次成功写入都会生成一个不可变的快照（snapshot），$snapshots 元数据表让你直观看到时间旅行（time travel）的基础——每个快照 ID 都可以作为后续读取的起点。更多读取方式（流式读取、指定快照 ID 增量读取等）见 Flink 查询文档。

使用内联配置创建表（Flink Connector 方式）

前面通过 CREATE CATALOG 定义 Flink Catalog 是一种方式；另一种方式是直接使用 Flink Connector，把 Catalog 连接信息内联写在表定义里。这种方式仍然连接同一个外部 Iceberg REST Catalog，区别仅在于无需单独的 CREATE CATALOG 语句。

CREATE TABLE taxis_inline_config (
vendor_id BIGINT,
trip_id BIGINT,
trip_distance FLOAT,
fare_amount DOUBLE,
store_and_fwd_flag STRING
) WITH (
'connector' = 'iceberg',
'catalog-name' = 'foo', — Required by Flink connector but value doesn't matter for inline config
'catalog-type' = 'rest',
'uri' = 'http://iceberg-rest:8181',
'warehouse' = 's3://warehouse/',
'io-impl' = 'org.apache.iceberg.aws.s3.S3FileIO',
's3.endpoint' = 'http://minio:9000',
's3.access-key-id' = 'admin',
's3.secret-access-key' = 'password',
's3.path-style-access' = 'true'
);

几点说明：

  • 这里的 Flink 表定义注册在 Flink 默认的内存 Catalog（default_catalog）中，但 Connector 属性告诉 Flink 将 Iceberg 表及其数据存放在与之前相同的 REST Catalog 与 S3 存储中；
  • catalog-name 是 Flink Connector 要求的必填项，但内联配置下其取值不影响连接目标；
  • 其余 uri、warehouse、io-impl、s3.* 参数与 CREATE CATALOG 场景完全一致，因为两者最终都是通过相同的 Catalog 配置来初始化连接（见 Flink Connector 文档 中 REST Catalog 一节）；
  • 同一套内联机制同样适用于 Hive、Hadoop 等 Catalog 类型，例如指定 'catalog-type'='hive' 加 uri 与 warehouse 即可。

关闭 Quickstart 环境

完成体验后，停止并移除所有容器：

docker compose -f docker/iceberg-flink-quickstart/docker-compose.yml down

如果想清理测试数据，可以先在 SQL Client 中执行 DROP TABLE iceberg_catalog.nyc.taxis; 与 DROP DATABASE iceberg_catalog.nyc;（参考 test.sql 的清理步骤）。

进一步学习

如果你希望在自建的 Flink 安装中包含 Iceberg 支持，可以把 Iceberg Flink runtime jar 添加到 Flink 的 jars（即 lib/）目录，运行时可从 Releases 页面 下载；Quickstart 镜像的 Dockerfile 展示了从 Maven Central 拉取 iceberg-flink-runtime、iceberg-aws-bundle 与 Hadoop bundle 的具体做法，可作参考。

至此你已经掌握了 Iceberg 在 Flink 下的最小实战闭环。下一步可以深入阅读 Iceberg Flink 文档 了解完整能力，包括流式读写（从指定快照增量读取）、DataStream API 读写、Flink 与 Iceberg 的 类型转换、以及 Flink 写入 / 查询 / 维护 等进阶主题。

赞

分享

  • 数据湖
  • 大数据
  • 数据存储

【免费下载链接】iceberg

Apache Iceberg

项目地址：
https://gitcode.com/gh_mirrors/icebe/iceberg

点击查看 免费下载

相关推荐

  • Daft 读写 AWS S3 Tables：基于 Catalog 接口与 Iceberg REST 的实战指南

    Daft 读写 AWS S3 Tables：基于 Catalog 接口与 Iceberg REST 的实战指南 本篇围绕 Daft 与 AWS S3 Table

    大数据
    数据分析
    数据工程
    AI 应用

  • SqlServerOnDocker：基于 Docker Compose 搭建 SQL Server 与 Django 开发环境的完整实践

    SqlServerOnDocker：基于 Docker Compose 搭建 SQL Server 与 Django 开发环境的完整实践 本指南基于 sql s

    示例工程
    数据库
    教程
    后端

  • 基于 docker-compose 搭建 SQL Server 快照复制环境：Publisher/Subscriber 容器化实操指南

    基于 docker compose 搭建 SQL Server 快照复制环境：Publisher/Subscriber 容器化实操指南 导读 本文以 sampl

    示例工程
    数据库
    教程
    后端

上一篇：
cJSON终极集成指南：5分钟搞定轻量级JSON解析器

下一篇：
no-mistakes递归门禁防护：gatecontext如何阻止自我嵌套

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

赞(0)
未经允许不得转载：171主机测评 » Apache Iceberg + Flink 快速上手实战：基于 Docker Compose 的 REST Catalog 环境搭建与 SQL 读写全流程
分享到： 更多 (0)

相关推荐

评论 抢沙发

取消

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

热门标签

#人工智能(64105)#python(21388)#开发语言(17795)#java(16368)#运维(16142)#linux(15622)#大数据(15291)#服务器(13643)#算法(12271)#前端(11627)#ai(10214)#网络(9888)#数据库(9487)#c++(8627)#学习(7154)#课程设计(7028)#机器学习(6997)#区块链(6054)#后端(5522)#OpenClaw(5412)#spring boot(5282)#深度学习(5260)#c语言(4546)#安全(4477)#毕业设计(4395)#数据结构(4337)#架构(4160)#docker(4109)#javascript(3972)#大模型(3903)

热门文章

最新文章

归档

分类