存储与表格式层

本层定位

传统数仓（如 Teradata、Greenplum）把存储与计算耦合在同一集群，扩容时“为算力买单却买回一堆磁盘”，缩容时数据要搬家。 湖仓一体的存储层通过“对象存储 + 开放表格式 + 元数据服务”三件套实现存算分离： 计算引擎无状态、可弹性伸缩，数据以开放格式长期沉淀在廉价可扩展的存储上。

与上层引擎解耦的原理

解耦不是口号，而是四条可验证的机制。逐层向下，每一层都只依赖“开放的协议”，而不依赖任何单一引擎：

1. 存算分离：引擎不持有数据，节点宕机/缩容不丢数据；同一份数据可同时被批、流、OLAP 引擎挂载。
2. 开放文件格式：数据落地为 Parquet 列存，这是行业公共协议，任何新引擎都能直接解析，无需“搬家”。
3. 自描述元数据：表结构、分区、快照全部记录在存储上的元数据文件中（如 Iceberg 的 metadata.json）， 引擎通过 Catalog 拿到入口指针即可独立读写，不依赖某个中心服务的运行状态。
4. 提交协议标准化：事务提交 = “原子地切换元数据指针”（条件写或 Catalog 乐观锁）， 避开对象存储“无原子 rename”的硬伤，多引擎并发写同一张表也不会互相覆盖。

Apache Iceberg 表格式

Apache Iceberg 由 Netflix 为解决“数 PB 级 Hive 表在日常运维中频繁出错”而设计，2020 年进入 Apache 顶级项目。 它在文件系统之上定义了一套树状元数据结构，为湖仓表带来四个核心能力：

• ACID 每次 commit 生成一个不可变快照（snapshot），读方永远看到一致视图；写方通过乐观并发 + 冲突重试实现串行化隔离。
• Schema 演进 安全地增列、删列、改名、调整列顺序，只改元数据、不重写数据文件；历史快照按旧 schema 依旧可读。
• 隐藏分区 分区值由列的变换函数（days / bucket / truncate）自动推导，用户与数据无需感知分区列。
• 时间旅行 任意历史快照可重复查询，天然支撑审计、回滚与增量消费。

元数据树：metadata.json → snapshot → manifest → data file

Iceberg 的一切能力都源于这棵“自描述”的元数据树。读写双方只需约定入口（当前 metadata.json），其余全部按文件指针逐级展开：

1. metadata.json：表入口文件，记录表 UUID、当前 schema、分区规格、属性与快照列表； 文件名自带版本号（00001-xxxx.metadata.json），提交即产生新版本。
2. snapshot：一次成功提交的不可变快照，包含操作类型（append / overwrite / delete）、 指向 manifest-list 的指针与汇总统计（added-files、total-records 等）。
3. manifest-list / manifest：清单文件（Avro）逐条记录数据文件路径、分区值与 min/max/null 统计，是分区裁剪与文件级跳过的物理依据。
4. data file：真正的 Parquet/ORC/Avro 数据文件，文件名含分布式唯一 ID，永不覆盖重写。

隐藏分区 vs Hive 分区

与 Delta Lake / Hudi 对比

Spark 建表：分区、排序与写入属性

-- 完整建表：隐藏分区 + 目标文件大小 + 压缩与元数据治理属性
CREATE TABLE lakehouse.dws_orders (
  order_id     BIGINT,
  user_id      STRING,
  city         STRING,
  order_amount DECIMAL(12, 2),
  order_ts     TIMESTAMP          -- 不需要冗余 dt 列，分区由变换推导
) USING iceberg
PARTITIONED BY (days(order_ts))               -- 隐藏分区：按天，自动裁剪
TBLPROPERTIES (
  'format-version'                     = '2',           -- v2 支持行级删除（MERGE/删除必需）
  'write.format.default'               = 'parquet',
  'write.target-file-size-bytes'       = '268435456',   -- 目标文件 256MB，防小文件
  'write.parquet.compression-codec'    = 'zstd',
  'write.parquet.row-group-size-bytes' = '134217728',   -- 行组 128MB
  'write.distribution-mode'            = 'hash',        -- 写入按分区键 shuffle，减少每任务小文件
  'write.metadata.previous-versions-max'        = '100',
  'write.metadata.delete-after-commit.enabled'  = 'true' -- 自动清理过期 metadata.json
);

-- 数据按时间聚簇（Z-Order/排序），提升范围扫描时文件级跳过命中率
ALTER TABLE lakehouse.dws_orders WRITE ORDERED BY (order_ts);

-- 需要按用户打散时可用多列排序，兼顾多种查询模式
ALTER TABLE lakehouse.dws_orders WRITE ORDERED BY (city, order_ts);

MERGE INTO：行级更新与打宽

-- 以订单号为主键：迟到更新覆盖、删除标记生效、新单插入，一次提交完成
MERGE INTO lakehouse.dws_orders t
USING (
  SELECT * FROM lakehouse.stg_orders
  WHERE etl_date = current_date()
) s
ON t.order_id = s.order_id
WHEN MATCHED AND s.op_type = 'D' THEN DELETE
WHEN MATCHED THEN UPDATE SET
  t.order_amount = s.order_amount,
  t.order_ts     = s.order_ts,
  t.city         = s.city
WHEN NOT MATCHED THEN INSERT (
  order_id, user_id, city, order_amount, order_ts
) VALUES (
  s.order_id, s.user_id, s.city, s.order_amount, s.order_ts
);

-- 执行计划验证：确认走了“重写受影响文件 + 删除文件”而非全表重写
EXPLAIN MERGE INTO lakehouse.dws_orders t USING lakehouse.stg_orders s
ON t.order_id = s.order_id WHEN MATCHED THEN UPDATE SET *;

时间旅行查询

-- 1）查看快照历史：拿到 snapshot_id 与提交时间
SELECT snapshot_id, committed_at, operation,
       summary['total-records'] AS records
FROM lakehouse.dws_orders.snapshots
ORDER BY committed_at DESC LIMIT 10;

-- 2）按 snapshot_id 精确回溯（VERSION AS OF）
SELECT count(*) FROM lakehouse.dws_orders VERSION AS OF 3448727692459611912;

-- 3）按时间点回溯（TIMESTAMP AS OF）：查“上周末对账时”的数据
SELECT * FROM lakehouse.dws_orders
TIMESTAMP AS OF '2026-11-01 00:00:00'
WHERE order_amount > 10000;

-- 4）增量消费：两个快照之间的变更（DataFrame API）
-- spark.read
--   .option("start-snapshot-id", "3448727692459611912")
--   .option("end-snapshot-id",   "7493257963288711204")
--   .table("lakehouse.dws_orders")

Flink SQL 写 Iceberg

-- Flink SQL：注册指向 Hive Metastore 的 Iceberg Catalog
CREATE CATALOG iceberg_hms WITH (
  'type'         = 'iceberg',
  'catalog-type' = 'hive',
  'uri'          = 'thrift://hms.lakehouse.local:9083',
  'warehouse'    = 's3a://lakehouse/warehouse',
  's3.endpoint'  = 'http://minio.lakehouse.local:9000',   -- MinIO 私有化部署时指定
  's3.path.style.access' = 'true'
);

-- 流式写入：数据的可见性由 Checkpoint 提交触发（详见 compute.html）
INSERT INTO iceberg_hms.lakehouse.dws_orders
SELECT order_id, user_id, city, order_amount, order_ts
FROM kafka_orders;

HDFS 分布式文件系统

尽管对象存储已成湖仓首选，HDFS 仍是大量存量私有机房的中坚。理解它的架构，也有助于理解对象存储“放弃”了哪些能力。

NameNode / DataNode 架构

• NameNode（主）：维护文件系统命名空间（目录树、块映射）于内存，持久化为 FSImage + EditLog； 是整个集群的单一元数据中心——也是小文件问题的根源（每个文件/目录/块约占数百字节内存）。
• DataNode（从）：存储实际数据块，定期心跳与块报告（BlockReport）给 NameNode，负责块的读写与副本流水线复制。
• 高可用：JournalNode 组成的共享 EditLog + Active/Standby 双 NameNode + ZKFC 自动故障切换， 避免单点瘫痪；SecondaryNameNode 只做镜像合并，不是热备。

块、副本与机架感知

文件被切分为固定大小的块（默认 128MB），每块默认 3 副本。副本放置由机架感知策略决定： 第 1 副本写客户端本节点（或随机），第 2 副本放到另一机架，第 3 副本与第 2 副本同机架不同节点—— 在“整个机架掉电”与“跨机架网络带宽”之间取得平衡。DataNode 启动时通过拓扑脚本上报机架位置：

<!-- core-site.xml：机架感知脚本（返回形如 /rack01 的机架路径） -->
<property>
  <name>net.topology.script.file.name</name>
  <value>/etc/hadoop/conf/rack-topology.sh</value>
</property>

关键参数

<!-- hdfs-site.xml：与湖仓负载最相关的参数 -->
<property>
  <name>dfs.replication</name>
  <value>3</value>                    <!-- 生产默认 3；归档冷表可降 2 或转纠删码 -->
</property>
<property>
  <name>dfs.blocksize</name>
  <value>268435456</value>            <!-- 256MB：湖仓大文件顺序扫描，减少块数与 NN 压力 -->
</property>
<property>
  <name>dfs.namenode.handler.count</name>
  <value>64</value>                   <!-- RPC 服务线程，重负载/多引擎并发时加大 -->
</property>
<property>
  <name>dfs.datanode.failed.volumes.tolerated</name>
  <value>1</value>                    <!-- 允许 1 块盘损坏不掉节点 -->
</property>
<property>
  <name>dfs.client.read.shortcircuit</name>
  <value>true</value>                <!-- 计算与存储同置时启用短路读，绕过 TCP -->
</property>

S3/MinIO 对象存储

对象存储把“文件系统”简化为“键值空间”：key 即路径，value 即不可变对象。它用极简的语义换取了近乎无限的扩展性与极低成本， 但也正是这层“简化”，逼出了 Iceberg 元数据协议的设计。

对象存储 vs HDFS

S3 的一致性短板与 Iceberg 的规避方式

经典的 Hive 提交协议是“写临时目录 → 原子 rename 到正式路径”。在 S3 上这条路径全面失效：

• rename = COPY + DELETE，非原子且昂贵（GB 级目录重命名要以小时计）；
• LIST 不是快照：分页枚举期间桶内在变化，无法作为“提交完成”的判定依据；
• 提交一半失败会留下“半张表”，读取方无从分辨。

Iceberg 的规避方式恰恰是把提交从“动数据”改为“动指针”：

1. 数据文件以分布式唯一 ID 命名，直接写入最终目录，永不重命名；文件不可变，先写后登记天然安全；
2. 提交时生成新的 manifest 与 snapshot，最后原子 PUT 一个新的 metadata.json （S3 单对象写是强一致的），或通过 HMS/JDBC/REST Catalog 的乐观锁切换指针；
3. 读取方从 Catalog 拿到确定的 metadata.json 入口，完全不需要 LIST 目录，也不依赖 rename；
4. 失败提交残留的“孤儿文件”由 remove_orphan_files 定期回收，不影响正确性。

S3A Commiter 配置

当 Spark 以原生 DataSource 方式（非 Iceberg 表）直接写 Parquet 到 S3 时，仍需 S3A 提交器来避免 “_temporary 目录 + rename 提交”陷阱。Iceberg 表自身不需要 rename，但集群级配置通常一并开启：

# spark-defaults.conf：magic committer（零 rename，通过隐藏目录 + _SUCCESS 标记提交）
spark.hadoop.fs.s3a.committer.name=magic
spark.hadoop.fs.s3a.committer.magic.enabled=true
spark.hadoop.mapreduce.outputcommitter.factory.scheme.s3a=org.apache.hadoop.fs.s3a.commit.magic.MagicCommitterFactory
spark.sql.sources.commitProtocolClass=org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol

# 或 directory committer（任务直接写目标目录，driver 最后汇总提交）
spark.hadoop.fs.s3a.committer.name=directory
spark.hadoop.fs.s3a.committer.directory.summary=true
spark.hadoop.mapreduce.outputcommitter.factory.scheme.s3a=org.apache.hadoop.fs.s3a.commit.directory.DirectoryCommitterFactory

# 大表写入吞吐：分块并行上传与连接池
spark.hadoop.fs.s3a.fast.upload=true
spark.hadoop.fs.s3a.threads.max=32
spark.hadoop.fs.s3a.connection.maximum=200
spark.hadoop.fs.s3a.block.size=64M          # 分块上传块大小

MinIO：私有化的 S3

MinIO 以 S3 兼容 API + 纠删码（Erasure Coding）+ Bitrot 校验著称，是私有机房/边缘环境落湖仓的首选底座。

# 单机体验：一条命令拉起对象存储 + Web 控制台
docker run -d --name minio \
  -p 9000:9000 -p 9001:9001 \
  -v /data/minio:/data \
  -e MINIO_ROOT_USER=lakehouse \
  -e MINIO_ROOT_PASSWORD='Lakehouse#2026' \
  minio/minio server /data --console-address ":9001"

# 生产分布式：4 节点 x 4 盘纠删码（分布式模式至少 4 节点）
# 在每个节点执行相同命令，主机名用 {1...4} 展开语法
export MINIO_ROOT_USER=lakehouse
export MINIO_ROOT_PASSWORD='Lakehouse#2026'
minio server http://minio{1...4}.lakehouse.local/data{1...4}

• 纠删码：默认 EC:4（4 数据 + 4 校验），空间利用率远高于 3 副本，可按桶调整；
• 部署要点：节点间时间同步（NTP）必须严格；生产建议 TLS 与独立域名；分布式模式扩容需按“纠删组”整组扩展；
• 定位：与公有云 S3 协议对齐后，开发测试用 MinIO、生产上 S3 的迁移基本只改 endpoint 与凭证。

Hive Metastore 元数据服务

湖仓多引擎的“表清单”需要一个中立裁判：Spark、Flink、Trino、Doris 各自的 catalog 互不相通， Hive Metastore（HMS）以 Thrift 服务的形式提供了跨引擎的共享目录——谁都能问它“库表清单、schema、存储位置”。

多引擎共享元数据

能力边界：HMS 只管“户口本”

关键配置

<!-- hive-site.xml：Metastore 服务端核心参数 -->
<property>
  <name>hive.metastore.uris</name>
  <value>thrift://hms1.lakehouse.local:9083,thrift://hms2.lakehouse.local:9083</value>
  <!-- 客户端可配多个地址轮询容灾 -->
</property>
<property>
  <name>hive.metastore.server.max.threads</name>
  <value>200</value>                <!-- 多引擎高并发时的 RPC 线程池 -->
</property>
<property>
  <name>hive.metastore.pool.max.active</name>
  <value>32</value>                 <!-- 底层 JDBC 连接池上限 -->
</property>
<property>
  <name>hive.metastore.metrics.enabled</name>
  <value>true</value>              <!-- 暴露 RPC/JMX 指标，接入监控 -->
</property>

HMS 瓶颈与解耦建议

• 瓶颈表现：所有引擎的 DDL 与首查都要过 HMS，高峰期 RPC 排队；Thrift 单实例能力有限， 底层数据库连接池打满后表现为“建表/查询 list 偶发超时”。
• 解耦路径一：Catalog 多实例：HMS 无状态化部署多实例（共享后端数据库）+ 负载均衡，按引擎划分访问组。
• 解耦路径二：换更轻的 Catalog：Iceberg 支持 JDBC Catalog（元数据表放 PostgreSQL/MySQL）、 REST Catalog（Nessie / Polaris / Gravitino 等），把“表清单”从 Hadoop 体系解放出来：

# Spark 挂载 Iceberg JDBC Catalog：不再依赖 HMS
spark.sql.catalog.lakehouse=org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.lakehouse.catalog-impl=org.apache.iceberg.jdbc.JdbcCatalog
spark.sql.catalog.lakehouse.uri=jdbc:postgresql://pg.lakehouse.local:5432/iceberg
spark.sql.catalog.lakehouse.jdbc.user=iceberg
spark.sql.catalog.lakehouse.jdbc.password=******
spark.sql.catalog.lakehouse.warehouse=s3a://lakehouse/warehouse

存储层调优要点

存储层的性能问题 90% 落在四个字上：小文件、坏布局。以下四项是投产前必须规划的内容。

1. 文件大小规划

2. Parquet 行组与压缩

-- 行组：列存的最小读取单元，建议 128MB–256MB，与扫描列宽匹配
ALTER TABLE lakehouse.dws_orders SET TBLPROPERTIES (
  'write.parquet.compression-codec'    = 'zstd',        -- 压缩比与解压速度的平衡点
  'write.parquet.compression-level'    = '3',           -- 1-22，越大越慢但更省
  'write.parquet.row-group-size-bytes' = '268435456',
  'write.parquet.page-size-bytes'      = '1048576'      -- 页是编码/统计最小单元
);

• 宽表点查场景（只取少数列）可适当减小行组，让“读到的无用数据”更少；
• snappy 解压最快、zstd 综合最优、gzip 只用于超冷归档；
• 查询侧确保 spark.sql.parquet.filterPushdown=true（默认开），让 min/max 谓词下推到行组。

3. 分区设计原则

1. 用天然高频过滤列：分区列必须出现在绝大多数查询的 WHERE 中，否则分区只是负担；
2. 基数适中：单表分区数建议控制在数千到数万级；按天分区足够就不要按小时；
3. 优先隐藏分区 transform：days(ts) / months(ts) / bucket(64, user_id)， 免冗余列、免手工指定；
4. 预估文件数：每日写入文件数 ≈ 分区数 × 每分区写入任务数，反推 target-file-size 是否合理。

4. 元数据爆炸治理

流式写入 = 高频提交 = 快照与 manifest 指数级累积。治理三件套必须进调度系统：

-- ① 快照过期：只保留最近 7 天 + 至少 100 个快照
CALL lakehouse.system.expire_snapshots(
  table => 'lakehouse.dws_orders',
  older_than => TIMESTAMP '2026-10-29 00:00:00',
  retain_last => 100
);

-- ② manifest 合并：大量小 manifest 会拖慢 planning
CALL lakehouse.system.rewrite_manifests(
  table => 'lakehouse.dws_orders'
);

-- ③ 孤儿文件清理：回收失败提交残留（低峰期执行，需引擎独占写）
CALL lakehouse.system.remove_orphan_files(
  table => 'lakehouse.dws_orders',
  older_than => TIMESTAMP '2026-11-04 00:00:00'
);

配合表属性做“写入时自愈”，避免依赖人工：

ALTER TABLE lakehouse.dws_orders SET TBLPROPERTIES (
  'write.metadata.previous-versions-max'       = '100',   -- metadata.json 最多留 100 版
  'write.metadata.delete-after-commit.enabled' = 'true',  -- 提交后自动删过期入口文件
  'write.metadata.metrics.default'             = 'truncate(16)', -- 长字符串统计截断，瘦身 manifest
  'commit.manifest.target-size-bytes'          = '8388608',     -- manifest 自动合并目标 8MB
  'commit.manifest.min-count-to-merge'         = '100'
);

适用边界

湖仓表不是银弹。以下场景，传统数仓或 OLAP 引擎原生表依然是更优解：

一个实用的判断口诀：“给机器反复扫的，放湖仓；给人实时看的，放 OLAP；给业务改的，放 OLTP”。 湖仓与这些引擎之间再通过联邦查询（见查询层一章）打通，形成互补贴而非互替代的组合。