简介本资源为基于云原生架构的大数据 Lakehouse 服务架构设计源码面向大数据开发工程师、架构学习者及需要搭建数据湖分析平台的技术团队帮助理解数据湖与数据仓库融合场景下的工程实现。压缩包共546个文件约2.75MB以310个Java文件为核心业务代码辅以66个ts与49个tsx构建前端界面28个Scala支撑大数据处理逻辑另有scss样式、xml与yaml配置、dockerfile容器化文件及sql脚本等覆盖从后端服务到前端展示的完整链路。项目包含lakehouse-common、lakehouse-ui、lakehouse-api等模块分别承担通用封装、界面展示与数据交互职责并集成Maven构建、Docker容器化与GitLab CI持续集成配置兼容主流云厂商对象存储。目前已有395人学习下载适合作为云原生大数据架构设计的参考案例帮助读者掌握多语言协作、模块划分与部署流程的工程实践。1. 云原生 Lakehouse 服务架构到底在解决什么问题如果你手上已经有一套跑在 YARN 上的 Hive 数仓或者正在用 Spark 直连对象存储做离线分析那你大概率遇到过这几个场景凌晨的 ETL 任务把整个集群的 CPU 吃满白天的即席查询慢到分析师直接放弃元数据散在 Hive Metastore、文件清单和业务库里改一张表的口径要拉三个群确认存算耦合导致扩容只能整机加节点成本压不下来。Lakehouse 服务架构要解决的就是把这套「数据湖存原始、数仓管结构」的割裂状态收敛成一套统一的表格式加统一的服务层再把它整个搬到云原生底座上跑。这里的关键词是三个云原生负责弹性与隔离大数据处理负责计算引擎与调度Lakehouse 负责在对象存储之上补回事务、Schema 演进和时间旅行。源码层面要落地的不是某一个算法而是一组服务元数据服务、表格式读写层、计算引擎适配层、以及把它们串起来的编排与治理。适合谁看适合已经会写 Spark/Flink 作业、但对「怎么把湖仓做成一个可运维的服务」还没打通链路的工程师。下面按「先立架构、再动手、最后避坑」的顺序拆开讲。2. 从存算耦合到 Lakehouse 服务化架构分层与选型理由2.1 为什么不能直接把 Hive 表搬到对象存储上很多人以为 Lakehouse 就是「Hive 表 S3」把LOCATION指到对象存储就完事了。实际跑起来会发现三个硬伤。第一Hive 的元数据只记录分区目录不记录文件级统计查询规划阶段拿不到列级 min/max谓词下推基本失效。第二对象存储的rename不是原子操作Spark 的INSERT OVERWRITE在提交阶段一旦失败会留下半截数据读的时候直接报文件不存在。第三并发写入没有冲突检测两个作业同时写一个分区后提交的静默覆盖前一个数据对不上账还查不出原因。Lakehouse 表格式Iceberg / Hudi / Delta Lake 这一类补的正是这三块快照隔离、文件级统计、原子提交。它们把「一张表」从「一堆目录」变成「一份带版本号的元数据清单」读的时候先读清单再读数据文件。这一步是服务化的前提因为只有元数据可编程才能在上面做权限、血缘、缓存和审计。2.2 云原生底座上计算与存储怎么切分云原生的核心价值是「按需申请、用完释放」落到 Lakehouse 上就是计算引擎无状态化。常见做法是把 Spark/Flink 的 Driver 和 Executor 都跑在 K8s 上用原生资源调度替代 YARN 的固定队列。存储侧统一走对象存储本地盘只做 shuffle 和缓存。这样带来两个直接收益一是不同团队用不同的资源池互不抢占二是作业结束 Pod 销毁不再有常驻的常开节点烧钱。但切分不是免费的。对象存储的 LIST 操作延迟高元数据服务如果每次查询都去 LIST 目录规划阶段就会卡住。所以架构上必须有一层独立的元数据服务Catalog Service把表清单缓存在内存或 KV 里对象存储只作为最终数据落点。这一层是整套 Lakehouse 服务架构里最容易被低估、也最容易成为瓶颈的部分。2.3 最小可跑通的服务分层清单把上面的判断落成一张分层表方便对照自己的环境缺哪块层级职责常见实现是否必须自研元数据服务表/分区/快照清单、并发提交Iceberg REST Catalog、Hive Metastore 改造建议复用只做适配表格式读写层快照隔离、Schema 演进、时间旅行Iceberg / Hudi / Delta复用不重写计算引擎适配Spark/Flink/Trino 读写表各引擎原生 connector复用编排与调度作业依赖、重试、资源申请Airflow / DolphinScheduler K8s Operator按团队规模定治理层权限、血缘、审计、质量Ranger 自研血缘采集需要自研采集这张表的意思是真正需要你写源码的地方集中在「元数据服务适配」和「治理层采集」两块表格式和引擎本身不要碰。很多团队一上来就想自己实现一套表格式最后都卡在并发提交的正确性上这是血泪经验。3. 用 Iceberg REST Catalog 在本地跑通最小读写链路3.1 环境准备与依赖版本约束本地复现不需要 K8s用 Docker 起一个对象存储MinIO加一个 REST Catalog 就够。依赖上要注意Iceberg 的 REST Catalog 和 Spark 版本是强绑定的Spark 3.5 配 Iceberg 1.4.x 比较稳混用容易出现NoSuchMethodError。下面用docker-compose起基础服务。# docker-compose.yml version: 3.8 services: minio: image: minio/minio:latest command: server /data --console-address :9001 environment: MINIO_ROOT_USER: admin MINIO_ROOT_PASSWORD: password ports: - 9000:9000 - 9001:9001 iceberg-rest: image: tabulario/iceberg-rest:latest environment: CATALOG_WAREHOUSE: s3://warehouse/ CATALOG_IO__IMPL: org.apache.iceberg.aws.s3.S3FileIO AWS_ACCESS_KEY_ID: admin AWS_SECRET_ACCESS_KEY: password AWS_REGION: us-east-1 CATALOG_S3_ENDPOINT: http://minio:9000 ports: - 8181:8181 depends_on: - minio这段配置里CATALOG_WAREHOUSE决定表数据落在哪个桶CATALOG_IO__IMPL指定用 S3 兼容的文件 IOCATALOG_S3_ENDPOINT是关键——不写它Catalog 会去连真实的 AWS 端点本地直接超时。启动后访问http://localhost:9001建一个名为warehouse的桶否则第一次建表会报Bucket does not exist。3.2 Spark 侧接入 REST Catalog 的配置Spark 通过spark-defaults.conf或运行时参数接入 REST Catalog。推荐写成配置文件避免每次提交都带一长串参数。# spark-defaults.conf spark.sql.extensionsorg.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions spark.sql.catalog.lakehouseorg.apache.iceberg.spark.SparkCatalog spark.sql.catalog.lakehouse.typerest spark.sql.catalog.lakehouse.urihttp://localhost:8181 spark.sql.catalog.lakehouse.warehouses3://warehouse/ spark.sql.catalog.lakehouse.io-implorg.apache.iceberg.aws.s3.S3FileIO spark.sql.catalog.lakehouse.s3.endpointhttp://localhost:9000 spark.sql.catalog.lakehouse.s3.path-style-accesstrue spark.sql.defaultCataloglakehousetyperest告诉 Spark 走 REST 协议而不是 Hive Metastores3.path-style-accesstrue是 MinIO 必须的因为 MinIO 不支持虚拟主机风格的桶访问defaultCatalog设成 lakehouse 后建表就不用每次写全限定名。这几行配错任何一行报错信息都很隐晦比如path-style没开时会报UnknownHostException: warehouse.minio。3.3 建表、写入、时间旅行的最小验证配置就绪后用 Spark SQL 跑一遍完整链路验证快照隔离和时间旅行是否生效。-- 建一张带分区的表 CREATE TABLE lakehouse.db.orders ( order_id BIGINT, user_id BIGINT, amount DECIMAL(10,2), dt STRING ) USING iceberg PARTITIONED BY (dt); -- 第一批写入 INSERT INTO lakehouse.db.orders VALUES (1, 1001, 99.50, 2024-01-01), (2, 1002, 150.00, 2024-01-01); -- 第二批写入制造第二个快照 INSERT INTO lakehouse.db.orders VALUES (3, 1003, 200.00, 2024-01-02); -- 查看快照列表 SELECT snapshot_id, committed_at, operation FROM lakehouse.db.orders.snapshots; -- 时间旅行读第一个快照 SELECT * FROM lakehouse.db.orders VERSION AS OF 第一个snapshot_id;PARTITIONED BY (dt)让 Iceberg 按天组织文件查询时能直接裁剪分区。snapshots元数据表是 Iceberg 自带的不需要额外建它记录了每次提交的快照 ID 和操作类型。时间旅行用VERSION AS OF指定快照 ID也可以用TIMESTAMP AS OF指定时间点。验证成功的标志是第二次写入后用第一个快照 ID 查出来的结果只有两行说明快照隔离生效了。提示本地验证时把spark.sql.catalog.lakehouse.cache-enabled设为 false避免元数据缓存掩盖并发问题等验证通过再打开。4. 计算引擎适配与元数据服务的并发写入处理4.1 Spark 与 Flink 双引擎写同一张表的冲突点真实生产里离线用 Spark 批量写、实时用 Flink 流式写同一张 Iceberg 表是常见需求。两者冲突集中在提交阶段Iceberg 的提交是乐观锁两个作业基于同一个父快照生成新快照先提交的成功后提交的会检测到父快照已过期抛CommitFailedException。Spark 侧默认会重试Flink 侧需要显式配置重试策略。// Flink Iceberg Sink 的重试配置 IcebergSink.forRowData(stream) .tableLoader(tableLoader) .writeParallelism(4) .upsert(true) .set(write.format.default, parquet) .set(commit.retry.num-retries, 10) .set(commit.retry.min-wait-ms, 500) .build();commit.retry.num-retries设成 10 是经验值太小在高并发下频繁失败太大则故障时提交延迟拉长。commit.retry.min-wait-ms控制退避起点配合指数退避能有效错开提交窗口。upsert(true)开启主键去重适合 CDC 场景但会带来额外的读放大纯追加场景不要开。4.2 元数据服务的缓存与一致性取舍REST Catalog 在高频查询下会成为瓶颈因为每次loadTable都要走一次 HTTP 加一次元数据反序列化。常见优化是在客户端加一层短 TTL 缓存但缓存会带来一致性问题A 作业刚提交的新快照B 作业可能因为缓存读到旧清单。取舍原则是——写路径不走缓存读路径可以容忍秒级延迟。# 客户端元数据缓存示意伪代码落在自研 Catalog 代理层 class CachedCatalog: def __init__(self, backend, ttl_seconds5): self.backend backend self.ttl ttl_seconds self.cache {} def load_table(self, name): entry self.cache.get(name) if entry and time.time() - entry.ts self.ttl: return entry.table table self.backend.load_table(name) self.cache[name] CacheEntry(table, time.time()) return table def commit_table(self, name, metadata): # 写路径直接穿透并主动失效缓存 result self.backend.commit_table(name, metadata) self.cache.pop(name, None) return result这段逻辑的关键在commit_table里主动pop缓存保证提交后下一次读能拿到新清单。TTL 设 5 秒是平衡点再长会导致下游看到过期数据再短缓存基本没意义。如果你的场景对一致性要求极高比如金融对账建议读路径也不缓存直接压后端用后端水平扩展扛。4.3 小文件合并的触发时机与参数流式写入 Iceberg 会产生大量小文件元数据清单膨胀查询规划变慢。Iceberg 提供rewrite_data_files动作做合并但触发时机很讲究太频繁会跟写入抢资源太久不合并查询会退化。-- 合并指定表的小文件目标文件大小 128MB CALL lakehouse.system.rewrite_data_files( table db.orders, options map( target-file-size-bytes, 134217728, min-input-files, 5, max-concurrent-file-group-rewrites, 10 ) );target-file-size-bytes设 128MB 是 Parquet 的常见甜点值跟 HDFS 块大小对齐。min-input-files设 5 表示一个文件组至少 5 个小文件才触发重写避免为了一两个文件白跑一趟。max-concurrent-file-group-rewrites控制并发设太大在共享集群上会挤占正常作业资源。建议把合并做成定时任务放在业务低峰期跑而不是每次写入后立即触发。5. 落地 Lakehouse 服务架构时最容易翻车的五个点5.1 现象作业偶发CommitFailedException重试后数据重复原因并发写入时乐观锁冲突Spark 默认重试会重新执行整个写入逻辑如果作业本身不是幂等的就会产生重复数据。解决写入前按主键做去重或者改用 Iceberg 的MERGE INTO替代INSERT让去重发生在表格式层而不是业务层。同时把commit.retry.num-retries调到合理值减少无谓重试。5.2 现象查询突然变慢snapshots表记录数暴涨原因流式写入没配小文件合并每个微批产生一个快照加一批小文件元数据清单线性增长。解决配置定时rewrite_data_files同时开启快照过期策略把不再需要的时间旅行快照清理掉。-- 清理 7 天前的快照保留最近 10 个 CALL lakehouse.system.expire_snapshots( table db.orders, older_than TIMESTAMP 2024-01-08 00:00:00, retain_last 10 );older_than和retain_last是双重保险前者按时间清后者保证至少留 10 个快照供回溯。注意清理快照会删除对应数据文件如果还有下游作业在读旧快照会直接报文件不存在所以清理前要确认没有长事务在读历史版本。5.3 现象对象存储费用比预期高出一截原因Iceberg 的expire_snapshots只删元数据引用底层文件如果没被正确标记为可删除或者用了不支持生命周期管理的存储类文件会一直留着。解决确认对象存储的垃圾回收策略跟 Iceberg 的删除动作对齐定期跑remove_orphan_files清理孤儿文件。CALL lakehouse.system.remove_orphan_files( table db.orders, older_than TIMESTAMP 2024-01-07 00:00:00 );older_than要设得比最长作业运行时间还长否则可能删掉正在写入但还没提交的文件这是最危险的操作之一务必留足缓冲。5.4 现象Schema 演进后老作业读取报字段错位原因Iceberg 支持加列、改列类型但老作业如果按位置解析 Parquet加列后字段顺序变化会导致错位。解决所有读写都按字段名而不是位置Parquet 读取时开启按名解析。同时 Schema 变更要走审批避免下游不知情。5.5 现象K8s 上 Executor 频繁被驱逐作业失败原因云原生环境下 Executor 是无状态 Pod节点资源紧张时会被优先驱逐Spark 默认不感知这种中断。解决给 Executor 配PodDisruptionBudget同时开启 Spark 的动态资源分配和任务重试让被驱逐的任务能在新 Pod 上重新调度。存储侧确保 shuffle 数据落在可靠的持久卷上否则 Pod 一挂 shuffle 数据就丢了。6. 用元数据表做血缘与成本归因的进阶玩法前面讲的都是「让链路跑通」这一章讲怎么让这套架构产生额外价值。Iceberg 的元数据表本身就是一份现成的血缘和成本数据源snapshots记录每次提交files记录每个数据文件的大小和分区manifests记录清单层级。把这些表定期同步到一张治理宽表里就能回答两个运维最关心的问题这张表的数据是谁写进来的以及它占了多少存储成本。具体做法是写一个定时采集作业把snapshots和files按表名聚合落到治理库。下面是一个采集片段# 采集 Iceberg 元数据表落到治理宽表 def collect_table_metrics(spark, table_name): snapshots spark.sql(f SELECT snapshot_id, committed_at, operation, summary FROM {table_name}.snapshots WHERE committed_at current_timestamp() - INTERVAL 1 DAY ) files spark.sql(f SELECT partition, file_path, file_size_in_bytes, record_count FROM {table_name}.files ) # 按分区聚合存储占用 storage files.groupBy(partition).agg( {file_size_in_bytes: sum, record_count: sum} ) return snapshots, storagesummary字段里带着added-records、added-files-size这类信息直接就能算出每次提交的增量成本。files表按分区聚合后能看出哪个分区是存储大户。把这两份数据按天快照存下来就得到了一条成本时间线比事后翻账单靠谱得多。验证这套采集是否准确有个简单办法拿某张表的files总大小跟对象存储控制台里对应前缀的实际占用对比差异应该在 5% 以内多出来的部分是孤儿文件或未过期快照。如果差异很大说明有文件没被正确追踪回头查remove_orphan_files的执行记录。我自己的习惯是每上一套新的 Lakehouse 表第一件事不是写业务查询而是先把元数据采集接上跑一周看成本曲线。这个习惯帮我提前发现过好几次小文件失控和快照泄漏比等账单出来再排查省事太多。希望帮到你。本文还有配套的精品资源点击获取
