4. Flink集成Iceberg环境搭建：版本兼容性、依赖引入与Catalog配置

环境搭建这件事，说白了就是先把地基打牢。我见过太多同学上来就写代码，结果跑起来全是ClassNotFound异常，一查发现是Flink和Iceberg版本没对齐。嗯，咱们今天就把这事彻底理清楚。

4.1 版本兼容性：选错版本，一天白干

Flink和Iceberg的版本匹配，是我在项目中踩过最深的坑之一。曾经有一次，我用了Flink 1.14和Iceberg 0.13，结果写Hive Catalog的时候一直报序列化错误，折腾了大半天才发现是版本不兼容。

下面这张表是我整理出来的稳定组合，你照着选基本不会出问题：

4.2 依赖引入：别漏了，也别多塞

依赖管理这块，我个人习惯用Maven做统一管理。你想想看，如果每个任务都手动丢jar包，维护起来就是噩梦。

核心依赖就这几样：

<!-- Flink 相关 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>

<!-- Iceberg Flink 集成 -->
<dependency>
    <groupId>org.apache.iceberg</groupId>
    <artifactId>iceberg-flink-runtime-1.15</artifactId>
    <version>${iceberg.version}</version>
</dependency>

<!-- Hive 依赖（如果用 Hive Catalog） -->
<dependency>
    <groupId>org.apache.hive</groupId>
    <artifactId>hive-metastore</artifactId>
    <version>${hive.version}</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<!-- Hadoop 配置（Hadoop Catalog 必备） -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>${hadoop.version}</version>
</dependency>

如果你是在SQL Client里跑，那就把对应的jar包丢到$FLINK_HOME/lib目录下。我个人更推荐用ADD JAR命令动态加载，这样不同任务可以用不同版本的Iceberg，互不干扰。

4.3 Catalog配置：两种方式，各有千秋

Catalog是Iceberg的元数据管理中心。说白了，就是告诉Flink「我的表存在哪里、怎么找到它」。Iceberg支持好几种Catalog，但生产上最常用的就两个：Hive Catalog和Hadoop Catalog。

4.3.1 Hive Catalog：企业级标配

如果你的公司已经有Hive Metastore，那Hive Catalog就是最自然的选择。它把Iceberg的元数据存在HMS里，这样Hive、Spark、Flink都能共享同一张表。

配置方式如下：

-- SQL Client 或 Flink SQL 中执行
CREATE CATALOG hive_catalog WITH (
    'type' = 'iceberg',
    'catalog-type' = 'hive',
    'uri' = 'thrift://hive-metastore:9083',
    'clients' = '5',
    'property-version' = '1',
    'warehouse' = 'hdfs://namenode:8020/user/iceberg/warehouse'
);

USE CATALOG hive_catalog;

这里有几个参数我要特别说一下：

• uri：Hive Metastore的Thrift地址。别写成HiveServer2的JDBC地址，我见过有人搞混。
• clients：客户端连接池大小。默认是2，但并发高的时候建议调到5-10。我在一次大促活动前忘了调这个，结果Metastore连接被打满，整个实时链路都堵了。
• warehouse：数据文件存放的根目录。这个路径要确保Flink任务有读写权限。

4.3.2 Hadoop Catalog：轻量级选择

Hadoop Catalog不需要依赖Hive Metastore，它直接把元数据以文件形式存在HDFS上。适合测试环境或者没有Hive的场景。

CREATE CATALOG hadoop_catalog WITH (
    'type' = 'iceberg',
    'catalog-type' = 'hadoop',
    'warehouse' = 'hdfs://namenode:8020/user/iceberg/hadoop_warehouse'
);

USE CATALOG hadoop_catalog;

4.4 整体架构：一张图看懂

下面这张图展示了Flink、Iceberg、Catalog和数据存储之间的关系。我画的时候特意把两种Catalog都放了进去，方便你对比理解。

4.5 验证环境：跑通第一个Iceberg表

环境搭好之后，我习惯先跑一个简单的DDL验证一下。你可以在Flink SQL Client里执行：

-- 创建数据库
CREATE DATABASE IF NOT EXISTS demo_db;

-- 创建Iceberg表
CREATE TABLE IF NOT EXISTS demo_db.user_behavior (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    ts TIMESTAMP(3),
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
    'format' = 'parquet',
    'write.format.default' = 'parquet',
    'write.target-file-size-bytes' = '134217728'
);

-- 插入一条数据测试
INSERT INTO demo_db.user_behavior VALUES 
    (1001, 2001, 'click', TIMESTAMP '2024-01-01 10:00:00');

-- 查询验证
SELECT * FROM demo_db.user_behavior;

嗯，环境搭建这部分就这些了。版本别选错，依赖别漏掉，Catalog按需配。这三件事做好了，后面写代码就是水到渠成的事。