数据湖与ETL：不止是存储，更是治理

聊到数据湖，很多刚入行的朋友第一反应是：「哦，就是个大池子，啥数据都能往里扔。」

嗯，这话对了一半。数据湖确实能存各种格式的数据——结构化、半结构化、非结构化，来者不拒。但如果你真把它当垃圾堆用，后面ETL的坑会让你哭都哭不出来。我见过太多项目，数据湖建起来容易，用起来却成了「数据沼泽」。

说白了，数据湖的核心不是存储，而是治理。今天我们就聊聊数据湖的底层逻辑，以及三个绕不开的利器：Delta Lake、Apache Iceberg、Apache Hudi。

数据湖的「三座大山」

传统数据仓库时代，我们得先建模再灌数据。数据湖反过来了——先存数据，需要时再定义结构。听起来很灵活，对吧？但实际落地时，你会遇到三个头疼的问题：

• ACID事务：多个任务同时读写同一份数据，怎么保证一致性？
• Schema演化：上游表突然加了个字段，下游ETL会不会崩？
• 时间旅行：我想查三天前的数据状态，能行吗？

这三个问题，传统Hive/Spark根本搞不定。于是Delta Lake、Iceberg、Hudi应运而生。它们不是存储引擎，而是表格式（Table Format）——在数据文件之上加了一层元数据管理层。

Delta Lake：Databricks的亲儿子

Delta Lake是我个人用得最多的方案。它基于Parquet文件，额外维护一个事务日志（Transaction Log）。每次读写操作，都会在日志里记一笔。

举个例子，你跑一个ETL任务往Delta表里写数据：

# 用Spark写Delta表
df.write.format("delta").mode("overwrite").save("/data/sales")

# 读历史版本
spark.read.format("delta").option("versionAsOf", 3).load("/data/sales")

这里有个细节：versionAsOf就是时间旅行。我在项目中排查数据问题时，经常靠这个回溯到某个版本，对比数据差异。没有它，你只能靠备份文件手动恢复，效率极低。

Apache Iceberg：开放生态的优等生

Iceberg的设计哲学和Delta Lake不太一样。它把元数据分成三层：元数据文件（Metadata）→ 清单列表（Manifest List）→ 清单文件（Manifest）→ 数据文件（Data File）。这种分层的好处是——查询时可以快速剪枝，只扫描需要的分区。

我参与过一个数据湖迁移项目，从Hive表切到Iceberg表。最让我惊艳的是它的隐藏分区（Hidden Partitioning）功能。你不需要手动指定分区字段，Iceberg会根据数据自动推断。比如：

-- 创建Iceberg表，自动按日期分区
CREATE TABLE iceberg_db.sales (
    order_id BIGINT,
    order_date DATE,
    amount DECIMAL(10,2)
) USING iceberg
PARTITIONED BY (days(order_date));

注意这里的days(order_date)，它是个转换函数。Iceberg会在写入时自动计算分区值，你完全不用操心。这在ETL流程里太省心了——少写一堆分区逻辑代码。

Apache Hudi：增量处理的急先锋

Hudi最擅长的是增量拉取（Incremental Pull）。你想想看，传统ETL里，我们经常要全量扫描一张大表，只为了获取最近几分钟的新数据。Hudi直接解决了这个痛点——它维护了提交时间线（Commit Timeline），你可以指定从某个时间点之后读取变更数据。

我曾在实时数仓项目里用Hudi做CDC（变更数据捕获）。上游MySQL的binlog通过Canal打到Kafka，然后Flink消费Kafka写入Hudi表。下游的报表任务只需要增量读取：

// 增量读取Hudi表
val hoodieIncView = spark.read.format("hudi")
    .option(DataSourceReadOptions.QUERY_TYPE_OPT_KEY, DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
    .option(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY, "20231001000000")
    .load("/data/hudi_orders")

这个BEGIN_INSTANTTIME就是关键。你不需要维护额外的偏移量，Hudi自己会记住每个批次的位置。嗯，这里要注意：增量查询的起始时间必须是提交时间线上存在的instant，否则会报错。

三者的对比与选择

说了这么多，到底选哪个？我整理了一张表，方便你对比：

特性	Delta Lake	Apache Iceberg	Apache Hudi
事务支持	强（基于事务日志）	强（基于快照隔离）	强（基于时间线）
Schema演化	支持（添加/删除/重命名列）	支持（更灵活的类型转换）	支持（需谨慎操作）
时间旅行	✅ 版本号或时间戳	✅ 快照ID或时间戳	✅ 时间戳
增量查询	需配合Change Data Feed	需配合Incremental Read	原生支持（最强）
生态集成	Spark为主，Flink有限	Spark/Flink/Trino等广泛	Spark/Flink/Hive
适用场景	Databricks环境、批处理为主	开放架构、多引擎查询	增量ETL、实时数仓

我个人习惯这样选：如果团队主要用Spark且部署在Databricks上，无脑选Delta Lake。如果追求开放生态，希望用Trino或Flink也能查，Iceberg更合适。如果业务对增量拉取有强需求，比如实时报表或CDC，Hudi是首选。

数据湖ETL实践：一个完整的例子

光说不练假把式。我们用一个实际场景串起来：假设你每天要从业务库同步订单数据到数据湖，然后做清洗和聚合。

第一步，用Spark Streaming消费Kafka中的订单消息，写入Iceberg表：

val df = spark.readStream.format("kafka")
    .option("subscribe", "orders_topic")
    .load()

df.selectExpr("CAST(value AS STRING) as json")
    .select(from_json($"json", schema).as("data"))
    .select("data.*")
    .writeStream
    .format("iceberg")
    .option("checkpointLocation", "/checkpoint/orders")
    .toTable("lake.orders")

第二步，每天凌晨跑一个批处理任务，做数据聚合：

spark.sql("""
    MERGE INTO lake.daily_summary t
    USING (
        SELECT order_date, product_id, SUM(amount) as total
        FROM lake.orders
        WHERE order_date = current_date() - 1
        GROUP BY order_date, product_id
    ) s
    ON t.order_date = s.order_date AND t.product_id = s.product_id
    WHEN MATCHED THEN UPDATE SET total = s.total
    WHEN NOT MATCHED THEN INSERT *
""")

这里用了MERGE INTO，也就是Upsert语义。Iceberg原生支持这个操作，不会产生重复数据。我在项目中用这个模式替代了传统的「先删后插」，性能提升了一倍不止。

数据湖的未来：湖仓一体

现在大家都在提「湖仓一体」（Lakehouse）。说白了，就是把数据湖的灵活性和数据仓库的可靠性结合起来。Delta Lake、Iceberg、Hudi就是湖仓一体的基石。它们让数据湖不再是「只写不读」的存储，而是真正能支撑BI报表、机器学习、实时分析的统一平台。

嗯，最后说一句：技术选型没有银弹。别盲目追新，先搞清楚你的业务场景是批处理为主还是增量为主，再决定用哪个。数据湖的ETL，说到底还是「数据治理」四个字——管好了，湖就是宝库；管不好，湖就是沼泽。

---