第16章数据湖仓一体（Lakehouse）¶

📚 章节概述¶

Lakehouse架构融合数据湖的灵活性与数据仓库的可靠性，在统一存储层上同时支持BI分析和AI/ML工作负载。

🎯 学习目标¶

完成本章后，你将能够：

理解Lakehouse架构的演进与核心特性
掌握Apache Iceberg和Delta Lake两大开放表格式
了解元数据管理与Catalog方案
掌握Time Travel与数据版本管理
能够设计生产级的Lakehouse架构

目录¶

1. Lakehouse架构概述
2. 开放表格式
3. Apache Iceberg深度解析
4. Delta Lake深度解析
5. Iceberg vs Delta Lake对比
6. 元数据管理
7. Time Travel与数据版本管理
8. 与Spark集成实战
9. Lakehouse最佳实践
10. 面试题精选
11. 推荐资源

1. Lakehouse架构概述¶

1.1 数据架构演进¶

Text Only

第一代：数据仓库 (Data Warehouse)
┌──────────┐    ETL    ┌───────────┐
│ 业务系统  │ ───────→ │  数据仓库   │ → BI报表
└──────────┘           └───────────┘
优点：ACID、Schema约束、查询性能好
缺点：成本高、不支持非结构化数据、扩展性差

第二代：数据湖 (Data Lake)
┌──────────┐   Ingest  ┌──────────┐
│ 业务系统  │ ───────→ │  数据湖    │ → AI/ML
│ 日志/IoT  │          │ (S3/HDFS) │
└──────────┘           └──────────┘
优点：低成本、支持所有数据格式、高扩展性
缺点：无ACID、数据沼泽、查询性能差

第三代：Lakehouse
┌──────────┐   Ingest  ┌──────────────────┐
│ 业务系统  │ ───────→ │   Lakehouse       │ → BI + AI
│ 日志/IoT  │          │ ┌──────────────┐  │
│ 流数据    │          │ │ 开放表格式层  │  │
└──────────┘           │ │(Iceberg/Delta)│  │
                       │ ├──────────────┤  │
                       │ │ 对象存储层    │  │
                       │ │(S3/ADLS/GCS) │  │
                       │ └──────────────┘  │
                       └──────────────────┘
优点：融合优势——ACID+低成本+所有数据+AI/BI统一

1.2 Lakehouse核心特性¶

特性	说明
ACID事务	支持原子性、一致性、隔离性、持久性
Schema Enforcement	写入时强制schema约束，防止数据质量问题
Schema Evolution	支持安全的schema变更（加列、改名等）
Time Travel	查询历史版本数据、支持回滚
开放格式	基于Parquet + 元数据层，避免厂商锁定
统一存储	BI查询和ML训练使用同一份数据
支持流批一体	同时支持批处理和流处理

1.3 Lakehouse典型架构¶

Text Only

         ┌──────────────────────────────────────────┐
查询引擎  │  Spark    Trino/Presto   Flink   DuckDB  │
         └──────────────────┬───────────────────────┘
                            │
         ┌──────────────────▼───────────────────────┐
Catalog  │    Hive Metastore / Iceberg REST Catalog  │
         │    Unity Catalog / AWS Glue Catalog        │
         └──────────────────┬───────────────────────┘
                            │
         ┌──────────────────▼───────────────────────┐
表格式    │   Apache Iceberg / Delta Lake / Hudi       │
         └──────────────────┬───────────────────────┘
                            │
         ┌──────────────────▼───────────────────────┐
文件格式  │    Apache Parquet / ORC / Avro             │
         └──────────────────┬───────────────────────┘
                            │
         ┌──────────────────▼───────────────────────┐
存储层    │    S3 / ADLS / GCS / HDFS / MinIO          │
         └──────────────────────────────────────────┘

2. 开放表格式¶

2.1 为什么需要表格式¶

原始Parquet文件存在的问题：

数据文件只有列式存储，无事务保证
无法安全地并发读写
没有schema演化能力
无法进行高效的小文件合并
无法查询历史版本

表格式 = Parquet数据文件 + 元数据层（事务日志/快照/Manifest）

2.2 三大开放表格式¶

维度	Apache Iceberg	Delta Lake	Apache Hudi
创始方	Netflix → Apache	Databricks	Uber → Apache
底层格式	Parquet/ORC/Avro	Parquet	Parquet/ORC
元数据	Manifest + Snapshot	Transaction Log（JSON）	Timeline
社区活跃度	★★★★★	★★★★☆	★★★☆☆
引擎兼容性	最广泛	Spark为主	Spark/Flink
核心优势	开放标准、多引擎	Databricks深度集成	Upsert性能好

3. Apache Iceberg深度解析¶

3.1 Iceberg架构¶

Text Only

                    ┌───────────────┐
    Catalog层       │  Catalog      │  存储表的当前元数据指针
                    │ (Hive/REST)   │
                    └───────┬───────┘
                            │
                    ┌───────▼───────┐
    元数据层        │  Metadata File │  表的schema、分区、属性
                    │  (.json)      │
                    └───────┬───────┘
                            │
                    ┌───────▼───────┐
    快照层          │  Snapshot      │  表在某个时间点的完整视图
                    │               │
                    └───────┬───────┘
                            │
                    ┌───────▼───────┐
    Manifest列表    │ Manifest List  │  指向一组Manifest的索引
                    │ (.avro)       │
                    └───────┬───────┘
                            │
                ┌───────────┼────────────┐
        ┌───────▼──────┐         ┌───────▼──────┐
        │ Manifest File│         │ Manifest File│  记录数据文件的元信息
        │ (.avro)      │         │ (.avro)      │  （分区值、统计信息）
        └───────┬──────┘         └───────┬──────┘
                │                        │
     ┌──────────┼──────┐      ┌──────────┼──────┐
     │          │      │      │          │      │
   data1     data2   data3  data4     data5   data6
  .parquet  .parquet .parquet .parquet .parquet .parquet

3.2 Iceberg核心特性¶

SQL

-- 创建Iceberg表
CREATE TABLE catalog.db.events (
    event_id    BIGINT,
    event_type  STRING,
    user_id     BIGINT,
    event_time  TIMESTAMP,
    properties  MAP<STRING, STRING>
)
USING iceberg
PARTITIONED BY (days(event_time), bucket(16, user_id))
TBLPROPERTIES (
    'write.format.default' = 'parquet',
    'write.parquet.compression-codec' = 'zstd'
);

-- Hidden Partitioning（隐式分区，查询无需感知分区列）
SELECT * FROM catalog.db.events
WHERE event_time >= '2025-01-01'
  AND event_time < '2025-02-01';
-- Iceberg自动进行分区裁剪，用户无需指定分区列

-- Schema Evolution
ALTER TABLE catalog.db.events ADD COLUMN device STRING;
ALTER TABLE catalog.db.events RENAME COLUMN properties TO metadata;
ALTER TABLE catalog.db.events ALTER COLUMN user_id TYPE BIGINT;

-- Partition Evolution（分区演化，无需重写数据）
ALTER TABLE catalog.db.events
  ADD PARTITION FIELD bucket(32, user_id);  -- 从16桶变为32桶

3.3 Iceberg分区裁剪¶

Text Only

传统Hive分区: 查询必须指定分区列
  WHERE dt = '2025-01-15'                ✓ 裁剪生效
  WHERE ts > '2025-01-15 10:00:00'       ✗ 全表扫描

Iceberg隐式分区: 自动推导分区值
  WHERE ts > '2025-01-15 10:00:00'       ✓ 自动裁剪days(ts)分区
  无需暴露分区实现细节给用户

4. Delta Lake深度解析¶

4.1 Delta Lake架构¶

Text Only

Delta Table = Parquet文件 + _delta_log/

table_path/
├── _delta_log/                    # 事务日志
│   ├── 00000000000000000000.json  # 版本0
│   ├── 00000000000000000001.json  # 版本1
│   ├── 00000000000000000002.json  # 版本2
│   └── 00000000000000000010.checkpoint.parquet  # 检查点
├── part-00000-xxx.snappy.parquet  # 数据文件
├── part-00001-xxx.snappy.parquet
└── part-00002-xxx.snappy.parquet

4.2 Delta Lake核心特性¶

Python

from delta import DeltaTable
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# 写入Delta表
df.write.format("delta").mode("overwrite").save("/data/events")

# MERGE操作（Upsert）
delta_table = DeltaTable.forPath(spark, "/data/events")

delta_table.alias("target").merge(
    updates_df.alias("source"),
    "target.event_id = source.event_id"
).whenMatchedUpdate(set={
    "event_type": "source.event_type",
    "properties": "source.properties"
}).whenNotMatchedInsertAll().execute()

# Change Data Feed（CDC）
spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 5) \
    .load("/data/events")

# OPTIMIZE（小文件合并）
delta_table.optimize().executeCompaction()

# Z-ORDER（列级聚类索引）
delta_table.optimize().executeZOrderBy("user_id", "event_time")

4.3 Delta Lake Liquid Clustering（新特性）¶

SQL

-- Liquid Clustering替代传统分区和Z-ORDER
CREATE TABLE events (
    event_id BIGINT,
    user_id BIGINT,
    event_time TIMESTAMP
) USING delta
CLUSTER BY (user_id, event_time);

-- 自动优化聚类
OPTIMIZE events;

5. Iceberg vs Delta Lake对比¶

5.1 详细对比¶

维度	Apache Iceberg	Delta Lake
元数据	多级Manifest树，高效裁剪	JSON事务日志，需定期Checkpoint
引擎支持	Spark/Flink/Trino/Presto/Dremio/StarRocks	Spark为主，Trino/Flink逐步支持
隐式分区	✅ 原生支持	❌ 需显式分区列
分区演化	✅ 无需重写数据	❌ 需重写
Merge性能	一般（Copy-on-Write为主）	优秀（支持Merge-on-Read）
行级更新	Position Delete / Equality Delete	Deletion Vector
Catalog	REST/Hive/Glue/Nessie	Unity Catalog/Hive
社区治理	Apache基金会（厂商中立）	Databricks主导（已开源）
生态集成	AWS/GCP/阿里云广泛支持	Databricks生态最佳

5.2 选型建议¶

Text Only

选Apache Iceberg:
├── 多引擎环境（Spark + Trino + Flink混用）
├── 需要厂商中立的开放标准
├── 需要隐式分区和分区演化
├── 大规模表（百万级文件）的元数据性能
└── 非Databricks平台

选Delta Lake:
├── 已使用Databricks平台
├── 需要优秀的Merge/Upsert性能
├── CDC场景较多
├── Spark为主要引擎
└── 需要Liquid Clustering等最新特性

6. 元数据管理¶

6.1 Catalog的作用¶

Text Only

Catalog = 表的注册中心
├── 存储表的位置信息
├── 管理表的schema
├── 提供表级访问控制
└── 支持多引擎发现和访问同一张表

6.2 常见Catalog方案¶

Catalog	类型	适用场景
Hive Metastore	传统	与Hadoop生态兼容
AWS Glue Catalog	托管	AWS环境
Iceberg REST Catalog	开放标准	多引擎、多云环境
Unity Catalog	Databricks	Databricks统一治理
Polaris Catalog	Snowflake开源	Iceberg原生Catalog
Nessie	开源	类Git版本管理Catalog

6.3 数据治理集成¶

Text Only

Lakehouse数据治理:
┌─────────────────────────────────────────┐
│  访问控制:  行级过滤 / 列级脱敏           │
│  数据血缘:  OpenLineage / Marquez        │
│  数据发现:  DataHub / OpenMetadata        │
│  数据质量:  Great Expectations / dbt test │
│  合规审计:  操作日志 + Time Travel        │
└─────────────────────────────────────────┘

7. Time Travel与数据版本管理¶

7.1 Iceberg Time Travel¶

SQL

-- 查询历史快照
SELECT * FROM catalog.db.events
  FOR SYSTEM_TIME AS OF '2025-06-01 00:00:00';

-- 按快照ID查询
SELECT * FROM catalog.db.events
  FOR SYSTEM_VERSION AS OF 1234567890;

-- 查看快照历史
SELECT * FROM catalog.db.events.snapshots;

-- 查看历史记录
SELECT * FROM catalog.db.events.history;

-- 回滚到指定快照
CALL catalog.system.rollback_to_snapshot('db.events', 1234567890);

-- 设置快照过期策略
ALTER TABLE catalog.db.events
  SET TBLPROPERTIES ('history.expire.max-snapshot-age-ms' = '604800000');  -- 7天

-- 手动清理过期快照
CALL catalog.system.expire_snapshots('db.events', TIMESTAMP '2025-06-01 00:00:00');

7.2 Delta Lake Time Travel¶

Python

# 按版本查询
df = spark.read.format("delta").option("versionAsOf", 5).load("/data/events")

# 按时间戳查询
df = spark.read.format("delta") \
    .option("timestampAsOf", "2025-06-01") \
    .load("/data/events")

# 回滚
delta_table = DeltaTable.forPath(spark, "/data/events")
delta_table.restoreToVersion(5)
# 或
delta_table.restoreToTimestamp("2025-06-01")

# 查看历史
delta_table.history().show()

# VACUUM清理旧版本文件（默认保留7天）
delta_table.vacuum(168)  # 小时数

7.3 Time Travel应用场景¶

场景	说明
数据审计	查看数据在某个时间点的状态
误操作回滚	恢复到错误操作之前的版本
可重复分析	对固定版本数据运行分析
增量处理	读取两个快照之间的变更
A/B测试	对同一数据集的不同版本运行对比

8. 与Spark集成实战¶

8.1 Spark + Iceberg¶

Python

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("IcebergDemo") \
    .config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.local.type", "hadoop") \
    .config("spark.sql.catalog.local.warehouse", "s3://my-bucket/warehouse") \
    .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.0") \
    .getOrCreate()

# 创建表
spark.sql("""
    CREATE TABLE local.db.events (
        event_id BIGINT,
        user_id BIGINT,
        event_type STRING,
        event_time TIMESTAMP
    ) USING iceberg
    PARTITIONED BY (days(event_time))
""")

# 写入数据
events_df.writeTo("local.db.events").append()

# Merge Into
spark.sql("""
    MERGE INTO local.db.events t
    USING updates s ON t.event_id = s.event_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

# 维护操作
spark.sql("CALL local.system.rewrite_data_files('db.events')")  # 小文件合并
spark.sql("CALL local.system.expire_snapshots('db.events', TIMESTAMP '2025-06-01')")

8.2 Spark Structured Streaming + Iceberg¶

Python

# 流式写入Iceberg表
streaming_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "events") \
    .load()

parsed_df = streaming_df.selectExpr(
    "CAST(key AS STRING) as event_id",
    "CAST(value AS STRING) as raw_data",
    "timestamp as event_time"
)

parsed_df.writeStream \
    .format("iceberg") \
    .outputMode("append") \
    .option("checkpointLocation", "s3://my-bucket/checkpoints/events") \
    .toTable("local.db.events")

# 流式读取Iceberg表（增量消费）
spark.readStream \
    .format("iceberg") \
    .option("stream-from-timestamp", "2025-06-01T00:00:00") \
    .load("local.db.events")

8.3 性能优化¶

SQL

-- 1. 小文件合并（Compaction）
CALL local.system.rewrite_data_files(
    table => 'db.events',
    options => map('target-file-size-bytes', '134217728')  -- 128MB
);

-- 2. 排序优化
CALL local.system.rewrite_data_files(
    table => 'db.events',
    strategy => 'sort',
    sort_order => 'user_id ASC NULLS LAST, event_time DESC'
);

-- 3. 统计信息收集
ALTER TABLE local.db.events
  SET TBLPROPERTIES ('write.metadata.metrics.default' = 'full');

-- 4. Bloom Filter
ALTER TABLE local.db.events
  SET TBLPROPERTIES ('write.parquet.bloom-filter-enabled.column.event_id' = 'true');

9. Lakehouse最佳实践¶

9.1 分层设计（Medallion架构）¶

Text Only

┌────────────────────────────────────────────────────┐
│                 Medallion Architecture               │
│                                                      │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐        │
│  │  Bronze   │ → │  Silver   │ → │  Gold     │       │
│  │  (原始层)  │   │  (清洗层)  │   │  (聚合层)  │       │
│  │          │   │          │   │          │        │
│  │ 原始数据  │   │ 清洗去重   │   │ 业务指标   │        │
│  │ 全量保留  │   │ Schema标准 │   │ 聚合报表   │        │
│  │ Append   │   │ 数据质量   │   │ ML特征    │        │
│  └──────────┘   └──────────┘   └──────────┘        │
└────────────────────────────────────────────────────┘

9.2 核心设计原则¶

选择合适的表格式：多引擎选Iceberg，Databricks选Delta
合理分区策略：避免过度分区（每个分区至少128MB）
定期维护：小文件合并、快照过期、孤儿文件清理
统一Catalog：所有引擎通过同一Catalog访问数据
数据质量前移：在Bronze→Silver阶段做好数据验证

10. 面试题精选¶

Q1: Lakehouse相比数据湖的核心优势？¶

ACID事务：保证数据一致性，支持并发读写
Schema Enforcement：写入时校验，防止数据质量问题
Time Travel：查询历史版本，支持回滚
统一存储：BI分析和ML训练使用同一份数据，消除数据孤岛
高性能：列裁剪、分区裁剪、统计信息等优化

Q2: Iceberg的隐式分区有什么优势？¶

用户查询时不需要知道分区列，引擎自动裁剪
分区演化不需要重写数据
避免了Hive分区导致的"查询必须带分区条件"的限制
支持基于时间的year/month/day/hour和bucket/truncate等转换

Q3: Delta Lake的事务日志如何工作？¶

每次写入操作生成一个JSON日志文件（版本号递增）
日志记录：AddFile、RemoveFile、Metadata变更等操作
读取时重放日志构建当前表状态
Checkpoint（每10个版本）：将日志压缩为Parquet格式，加速读取
并发写入通过乐观并发控制解决冲突

Q4: Medallion架构的三层分别做什么？¶

层次	数据特征	处理逻辑
Bronze	原始、未处理	1:1复制源数据，保留全部字段
Silver	清洗、标准化	去重、类型转换、数据质量验证
Gold	聚合、业务化	指标计算、维度建模、特征工程

Q5: 如何处理Lakehouse中的小文件问题？¶

定期运行rewrite_data_files（Iceberg）或OPTIMIZE（Delta）
设置合理的目标文件大小（推荐128MB-256MB）
流式写入使用较大的触发间隔或上游攒批
配置自动优化（Auto-compaction）

11. 推荐资源¶

下一步学习：结合05-数据湖技术理解底层存储，参考17-dbt数据转换实现Lakehouse上的数据转换流水线。

第16章 数据湖仓一体（Lakehouse）¶

📚 章节概述¶

🎯 学习目标¶

目录¶

1. Lakehouse架构概述¶

1.1 数据架构演进¶

1.2 Lakehouse核心特性¶

1.3 Lakehouse典型架构¶

2. 开放表格式¶

2.1 为什么需要表格式¶

2.2 三大开放表格式¶

3. Apache Iceberg深度解析¶

3.1 Iceberg架构¶

3.2 Iceberg核心特性¶

3.3 Iceberg分区裁剪¶

4. Delta Lake深度解析¶

4.1 Delta Lake架构¶

4.2 Delta Lake核心特性¶

4.3 Delta Lake Liquid Clustering（新特性）¶

5. Iceberg vs Delta Lake对比¶

5.1 详细对比¶

5.2 选型建议¶

6. 元数据管理¶

6.1 Catalog的作用¶

6.2 常见Catalog方案¶

6.3 数据治理集成¶

7. Time Travel与数据版本管理¶

7.1 Iceberg Time Travel¶

7.2 Delta Lake Time Travel¶

7.3 Time Travel应用场景¶

8. 与Spark集成实战¶

8.1 Spark + Iceberg¶

8.2 Spark Structured Streaming + Iceberg¶

8.3 性能优化¶

9. Lakehouse最佳实践¶

9.1 分层设计（Medallion架构）¶

9.2 核心设计原则¶

10. 面试题精选¶

Q1: Lakehouse相比数据湖的核心优势？¶

Q2: Iceberg的隐式分区有什么优势？¶

Q3: Delta Lake的事务日志如何工作？¶

Q4: Medallion架构的三层分别做什么？¶

Q5: 如何处理Lakehouse中的小文件问题？¶

11. 推荐资源¶

第16章数据湖仓一体（Lakehouse）¶