2、快速上手：环境准备（Spark/Flink）、创建第一个Iceberg表、写入与读取数据、理解快照概念

好，咱们直接动手。这一章的目标很简单——让你在本地把 Iceberg 跑起来，亲眼看看它到底怎么工作的。我个人习惯是先用 Spark 做快速验证，因为它的交互式 shell 最适合教学。等跑通了，再用 Flink 感受一下流式写入。

嗯，这里要提前说一句：别被环境配置吓到。我当年第一次搭 Iceberg 环境时也踩了不少坑，尤其是依赖冲突。但只要你照着下面的步骤来，半小时内肯定能见到第一个 Iceberg 表。

2.1 环境准备：Spark + Iceberg 集成

我们先从 Spark 开始。Iceberg 官方提供了专门的 Spark 运行时 JAR，把依赖都打包好了，省心不少。

2.1.1 下载与配置

你需要准备三样东西：

• Apache Spark 3.4+（我推荐 3.5.1，兼容性最好）
• Iceberg Spark Runtime JAR（版本 1.5.0 以上）
• 本地文件系统或 HDFS（做练习用本地路径就行）

下载完 Spark 后，解压到任意目录。然后启动 Spark Shell 时，通过 --packages 参数引入 Iceberg：

./bin/spark-shell \
  --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.0 \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.demo=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.demo.type=hadoop \
  --conf spark.sql.catalog.demo.warehouse=/tmp/iceberg_warehouse

这里我解释一下几个配置的含义：

2.2 创建第一个 Iceberg 表

Spark Shell 启动成功后，你会看到 scala> 提示符。接下来我们创建一个最简单的用户行为表：

spark.sql("""
  CREATE TABLE demo.db.user_events (
    event_id BIGINT,
    user_id STRING,
    event_type STRING,
    event_time TIMESTAMP,
    event_data MAP<STRING, STRING>
  ) USING iceberg
  PARTITIONED BY (days(event_time))
""")

执行完后，去 /tmp/iceberg_warehouse 目录看看：

$ ls -l /tmp/iceberg_warehouse/db/user_events/
drwxr-xr-x  metadata/
drwxr-xr-x  data/

你会看到两个目录：metadata 和 data。这就是 Iceberg 的核心——元数据与数据文件分离。传统 Hive 表可没有这么清晰的划分。

2.3 写入数据

表建好了，我们来写点数据。Iceberg 支持标准的 INSERT INTO，也支持 DataFrame 写入。我建议你两种都试试：

2.3.1 SQL 方式写入

spark.sql("""
  INSERT INTO demo.db.user_events VALUES
    (1, 'u001', 'click', TIMESTAMP '2024-01-15 10:00:00', map('page', 'home')),
    (2, 'u002', 'view',  TIMESTAMP '2024-01-15 10:05:00', map('page', 'product')),
    (3, 'u001', 'purchase', TIMESTAMP '2024-01-15 10:10:00', map('amount', '99.9'))
""")

2.3.2 DataFrame 方式写入

import spark.implicits._

val df = Seq(
  (4L, "u003", "click", "2024-01-16 09:00:00", Map("page" -> "search")),
  (5L, "u004", "view",  "2024-01-16 09:15:00", Map("page" -> "cart"))
).toDF("event_id", "user_id", "event_type", "event_time", "event_data")
  .withColumn("event_time", $"event_time".cast("timestamp"))

df.writeTo("demo.db.user_events").append()

写入完成后，我们查一下数据：

spark.sql("SELECT * FROM demo.db.user_events").show()
+--------+-------+----------+-------------------+--------------------+
|event_id|user_id|event_type|         event_time|          event_data|
+--------+-------+----------+-------------------+--------------------+
|       1|   u001|     click|2024-01-15 10:00:00|        {page->home}|
|       2|   u002|      view|2024-01-15 10:05:00|     {page->product}|
|       3|   u001|  purchase|2024-01-15 10:10:00|    {amount->99.9}|
|       4|   u003|     click|2024-01-16 09:00:00|      {page->search}|
|       5|   u004|      view|2024-01-16 09:15:00|       {page->cart}|
+--------+-------+----------+-------------------+--------------------+

数据出来了，看起来和普通表没什么区别，对吧？别急，好戏在后面。

2.4 理解快照（Snapshot）概念

这才是 Iceberg 最核心的东西。每次对 Iceberg 表的写入、删除、更新操作，都会生成一个快照。你可以把快照理解为表在某个时间点的「完整视图」。

我们来看看刚才的操作产生了几个快照：

spark.sql("SELECT * FROM demo.db.user_events.snapshots").show(false)
+------------------------+-------------------+---------+-------------------------+
|committed_at            |snapshot_id        |parent_id|operation                |
+------------------------+-------------------+---------+-------------------------+
|2024-01-15 10:00:00...  |1234567890123456781|null     |append                   |
|2024-01-15 10:05:00...  |1234567890123456782|123456...|append                   |
|2024-01-16 09:00:00...  |1234567890123456783|123456...|append                   |
+------------------------+-------------------+---------+-------------------------+

看到了吗？三次写入，三个快照。每个快照都有一个唯一的 snapshot_id，并且记录了父快照 ID，形成一条链。

2.4.1 时间旅行查询

快照最大的价值就是时间旅行。你可以查询表在任意历史快照时的状态：

-- 查询第一个快照时的数据（只有前3条）
spark.sql("""
  SELECT * FROM demo.db.user_events
  FOR SYSTEM_VERSION AS OF 1234567890123456781
""").show()

-- 或者用时间戳
spark.sql("""
  SELECT * FROM demo.db.user_events
  FOR SYSTEM_TIME AS OF '2024-01-15 10:02:00'
""").show()

为什么会这样？因为 Iceberg 的元数据层记录了每个快照对应的文件列表。查询时，引擎只读取该快照指向的那些文件，其他文件完全忽略。

我曾在一次生产事故中靠这个功能救了场。同事误删了一批数据，我直接用时间旅行查到了删除前的快照，把数据捞了回来。你想想看，要是用传统 Hive 表，数据被覆盖了就真没了。

2.5 用 Flink 写入 Iceberg

Spark 搞定了，我们再来看看 Flink。Flink 的流式写入是 Iceberg 的另一大亮点。

2.5.1 Flink SQL Client 配置

启动 Flink SQL Client 时，同样需要引入 Iceberg 依赖：

./bin/sql-client.sh \
  -j /path/to/iceberg-flink-runtime-1.18-1.5.0.jar \
  -pyfs /path/to/flink-sql-connector-hive-3.1.3_2.12-1.18.0.jar

然后在 SQL Client 中创建 Catalog：

CREATE CATALOG demo_catalog WITH (
  'type' = 'iceberg',
  'catalog-type' = 'hadoop',
  'warehouse' = '/tmp/iceberg_warehouse'
);

USE CATALOG demo_catalog;

2.5.2 流式写入示例

假设我们有一个 Kafka 数据源，实时写入 Iceberg 表：

CREATE TABLE db.user_events_sink (
  event_id BIGINT,
  user_id STRING,
  event_type STRING,
  event_time TIMESTAMP(3),
  event_data MAP<STRING, STRING>
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = 'demo_catalog',
  'catalog-database' = 'db',
  'catalog-table' = 'user_events',
  'write.format.default' = 'parquet',
  'write.upsert.enabled' = 'false'
);

-- 从 Kafka 读取并写入 Iceberg
INSERT INTO db.user_events_sink
SELECT * FROM kafka_source;

Flink 会以 checkpoint 为边界，每个 checkpoint 生成一个 Iceberg 快照。这样即使作业失败，也能从最近成功的 checkpoint 恢复，保证 exactly-once 语义。

2.6 本章核心知识图

下面这张图总结了 Iceberg 快照的核心逻辑，你一看就明白：

从图中你能看到：快照 1 指向 data_file_1、2、3；快照 2 新增了 data_file_4；快照 3 又新增了 data_file_5。查询快照 2 时，Iceberg 只读取前 4 个文件，完全不知道 data_file_5 的存在。

2.7 本章小结

好了，这一章我们做了几件实事：

• 搭好了 Spark + Iceberg 环境
• 创建了第一个 Iceberg 表
• 用 SQL 和 DataFrame 两种方式写入了数据
• 理解了快照的概念，并实践了时间旅行查询
• 用 Flink 体验了流式写入

说白了，Iceberg 的核心就一句话：每次写入生成一个轻量级快照，快照指向一组数据文件，查询时可以自由选择任意快照。这个机制让数据版本管理变得异常简单。