迁移前的环境准备：Spark/Flink 版本选择、Catalog 配置、依赖包管理

说实话，很多团队在湖仓表格式迁移上栽跟头，不是因为迁移逻辑多复杂，而是环境准备阶段就埋了雷。我见过不止一次，项目都跑起来了，突然发现 Spark 版本和 Iceberg 不兼容，或者 Flink 的 Catalog 连不上 Hive Metastore。嗯，这章咱们就把这些坑提前填上。

2.1 计算引擎版本选择：别追新，要追稳

选版本这事，我的原则很简单：用社区验证过的组合，别当小白鼠。你想想看，迁移项目本身就有风险，再搭上一个刚发布两周的引擎版本，出了问题连 Stack Overflow 都搜不到答案。

2.1.1 Spark 版本选型

目前主流的数据湖表格式（Iceberg、Delta Lake、Hudi）对 Spark 3.x 支持最好。我个人习惯这样选：

2.1.2 Flink 版本选型

Flink 这边情况类似。如果你主要用 Flink 做实时写入或 CDC 入湖，版本搭配要更谨慎：

• Flink 1.16 / 1.17：目前最稳的选择，Iceberg 和 Hudi 的 Flink connector 都做了充分测试
• Flink 1.18：可以用了，但建议等 1.18.1 或 1.18.2
• Flink 1.15 及以下：除非你存量系统就是这版本，否则别新起项目用

2.2 Catalog 配置：打通元数据的任督二脉

Catalog 是什么？说白了就是表格式的"户口本"。没有它，Spark 和 Flink 根本不知道你的表存在哪里、有哪些分区、数据文件在哪。我见过最典型的翻车现场：Catalog 配好了但权限没开，结果跑任务时一直报 AccessDeniedException。

2.2.1 Hive Metastore 作为统一 Catalog

大多数企业已经有 Hive Metastore（HMS）了。复用 HMS 作为 Catalog 是最省事的方案。配置起来其实不复杂：

# Spark 侧配置（spark-defaults.conf 或 session 级别）
spark.sql.catalog.my_catalog        = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.my_catalog.type   = hive
spark.sql.catalog.my_catalog.uri    = thrift://hive-metastore:9083
spark.sql.catalog.my_catalog.warehouse = hdfs://namenode/user/hive/warehouse

# Flink 侧配置（flink-conf.yaml 或 Table API）
CREATE CATALOG my_catalog WITH (
  'type' = 'iceberg',
  'catalog-type' = 'hive',
  'uri' = 'thrift://hive-metastore:9083',
  'warehouse' = 'hdfs://namenode/user/hive/warehouse'
);

2.2.2 独立 Catalog（不依赖 HMS）

有些场景下，你可能不想碰 HMS。比如新项目、云原生环境、或者 HMS 太老旧改不动。这时候可以用独立 Catalog：

• Hadoop Catalog：直接基于文件系统，适合测试或小规模场景
• JDBC Catalog：用关系数据库存元数据，适合生产环境
• REST Catalog：Iceberg 0.14+ 推出的新方式，通过 REST API 交互，适合跨引擎共享

我个人比较推荐 JDBC Catalog 作为 HMS 的替代方案。配置也很直接：

# Spark 侧
spark.sql.catalog.my_catalog        = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.my_catalog.type   = jdbc
spark.sql.catalog.my_catalog.uri    = jdbc:postgresql://pg-host:5432/iceberg_catalog
spark.sql.catalog.my_catalog.jdbc.user     = iceberg_user
spark.sql.catalog.my_catalog.jdbc.password = your_password

2.3 依赖包管理：版本冲突是最大的敌人

依赖包管理，嗯，这可能是整个迁移准备阶段最让人头疼的部分。你想想看，Spark 自带一堆 Jackson、Guava、Netty，Flink 也带一堆，Iceberg 又依赖另一堆……版本稍微不对付，就是 NoSuchMethodError 或 ClassNotFoundException 满天飞。

2.3.1 核心依赖清单

以 Iceberg 为例，你需要准备这些 jar 包：

2.3.2 依赖冲突排查三板斧

我曾经在一个项目上花了两天时间排查依赖冲突，最后发现是 Guava 版本不一致。从那以后，我总结了一套排查流程：

1. 用 spark-submit --verbose 看 classpath：启动时加这个参数，Spark 会打印出所有加载的 jar 包路径
2. 用 mvn dependency:tree 分析传递依赖：在 pom.xml 里跑这个命令，能看到每个依赖的版本树
3. 用 Class.forName("com.google.common.base.Preconditions") 测试：写个简单的 Spark 脚本，加载有嫌疑的类，看是从哪个 jar 包来的

2.3.3 一个完整的依赖配置示例

下面是我在一个生产项目中实际用过的 Spark 提交命令，你可以参考：

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.my_catalog=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.my_catalog.type=hive \
  --conf spark.sql.catalog.my_catalog.uri=thrift://hms-host:9083 \
  --jars /opt/iceberg-libs/iceberg-spark-runtime-3.3_2.12-1.3.1.jar,\
        /opt/iceberg-libs/hive-metastore-3.1.3.jar,\
        /opt/iceberg-libs/hadoop-common-3.3.4.jar \
  your-migration-job.jar

2.4 环境验证：别等跑起来才发现问题

所有配置都搞定了？别急，先跑一个简单的验证脚本。我习惯用下面这个"三连验证"：

-- 1. 验证 Catalog 是否连通
SHOW NAMESPACES FROM my_catalog;

-- 2. 验证读写能力
CREATE TABLE my_catalog.test_db.test_table (
  id INT,
  name STRING,
  ts TIMESTAMP
) USING iceberg
PARTITIONED BY (days(ts));

INSERT INTO my_catalog.test_db.test_table VALUES (1, 'hello', current_timestamp());

SELECT * FROM my_catalog.test_db.test_table;

-- 3. 验证时间旅行（Iceberg 核心能力）
SELECT * FROM my_catalog.test_db.test_table FOR SYSTEM_TIME AS OF '2024-01-01 00:00:00';

2.5 本章知识体系总览

下面这张图把本章的核心逻辑串起来了，你可以对照着检查自己的环境准备是否完整：