极点宏观｜Amarathon 2025 回顾 33：构建流式 Iceberg 表以进行实时物流分析

现代物流面临的挑战

● 管理卡车、司机、路线、燃料、维护、货运和仓库等多个数据流。

● 需要实时运营视图和长期分析。

数据存储要求

● 提供最新的连接视图，以支持即时运营。

● 使用 Apache Iceberg 进行长期分析。

技术栈

● RisingWave：提供流处理能力的数据平台。

● Lakekeeper：用于数据管理的开放式 REST catalog。

● Kafka：流数据的事件骨干。

● 对象存储（例如 MinIO）：数据存储解决方案。

目标

● 演示如何使用指定的开放式技术栈构建流式 Iceberg 表。

● 为现代物流数据管理提供简单有效的解决方案。

物流分析问题

● 当今的物流平台会生成：

● 卡车：车队清单和位置

● 司机：名册和任务分配

● 货运：始发地、目的地和重量

● 仓库：容量和站点

● 路线：预计到达时间（ETA）和距离

● 燃料与维护：成本和可靠性信号

● 挑战：

● 运营团队需要涵盖所有这些数据流的最新连接视图。

● 数据团队需要将同一份数据存入 Iceberg，以进行 BI、AI 和历史分析。

我们将构建的内容（流式 Iceberg 模式）

● Kafka 将七个物流主题送入 RisingWave。

● 使用 SQL 表达多路流连接，并在 RisingWave 内持续进行物化。

● 结果从 RisingWave 持久化为原生 Apache Iceberg 表，并存储在 MinIO 等兼容 S3 的对象存储中。

● Spark、Trino 和 DuckDB 等引擎通过开放式 REST catalog 查询相同的 Iceberg 表。

为何使用 RisingWave 构建流式 Iceberg 表？

[ 1 ] 批处理优先的工作流：

● 周期性作业、过时的连接结果和繁重的数据流水线。

● 需要使用单独的 ETL 工具写入 Iceberg。

[ 2 ] RisingWave + 流式 Iceberg 表：

● 在 RisingWave MV 中持续更新连接和聚合结果。

● 始终“近乎实时”的 Iceberg 快照。

● 一条 RisingWave 数据流水线同时支持实时仪表盘和离线分析。

● 目标：由 RisingWave 负责流数据流水线和 Iceberg 写入，让 Iceberg 的使用体验如同数据库。

---

高层架构

● 我们的端到端技术栈：

● Kafka — 7 个物流主题的事件骨干。

● RisingWave（流数据库）— 使用 SQL 摄取、连接和聚合数据；管理物化视图。

● RisingWave Iceberg Table Engine + Lakekeeper — Iceberg 表的开放式 REST catalog。

● MinIO — 兼容 S3 的对象存储。

● 模式：Kafka → RisingWave → MinIO 中的 Iceberg → 任何引擎均可通过 REST catalog 查询。

RisingWave 中的物流数据流与多路流连接

● RisingWave 中的七个物流数据流

● 本示例使用七个 Kafka 主题，这些主题会成为 RisingWave 中的 source：

● trucks — 车队清单、容量、当前位置。

● driver — 司机详细信息和 assigned_truck_id。

● shipments — 始发地、目的地、重量、卡车绑定关系。

● warehouses — 仓库位置和容量。

● route — route_id、truck_id、driver_id、ETD/ETA、distance_km。

● fuel — 加油事件（时间、升数、加油站）。

● maint — 维护历史和成本。

● RisingWave 将每一个数据流都视为流式表，可直接使用简单的 PostgreSQL 风格 SQL 进行连接。

模式 1：RisingWave 中的多路流连接

● 在 RisingWave 中，我们将核心物流逻辑表达为一个多路流连接。

● LEFT JOIN drivers → trucks，以保留未匹配的司机数据。

● JOIN shipments，以附加工作负载和目的地。

● JOIN warehouses，以加入容量和位置信息。

● JOIN route，以获取 ETD/ETA 和距离。

● JOIN fuel 和 maint，以获取成本和可靠性信号。

● 这会生成 logistics_joined_mv — RisingWave 内针对每辆卡车/每位司机/每条路线持续更新的反规范化物流记录。

---

车队 KPI、原生 Iceberg 表与跨引擎读取 模式 2

RisingWave 中的车队 KPI 视图

● 在连接后的 MV 之上，我们另外定义一个用于车队 KPI 的 RisingWave MV：

● 每辆卡车的容量利用率（%）。

● 每辆卡车的燃料总成本和维护成本。

● 合计运营总成本。

● 当前路线信息（ID、ETD、ETA、distance_km）。

● 相关司机详细信息。

RisingWave 中的 overview 会成为实时车队绩效表 供 Grafana 和运营仪表盘使用。模式 3：从 RisingWave 流式写入原生 Iceberg

● 不使用自定义 writer 服务，而是：

● [ 1 ] 将 logistics_joined_iceberg 定义为由 RisingWave 管理的原生 Iceberg 表。

● [ 2 ] 其 schema 与 logistics_joined_mv 一致。

● [ 3 ] RisingWave 中的一小段配置可控制将流式变更提交为 Iceberg 快照的频率。

模式 4：通过 REST catalog 进行跨引擎读取

● 使用由 RisingWave 创建并注册到 Lakekeeper REST catalog 的 Iceberg 表：

● [ 1 ] Spark 将 lakekeeper 挂载为 catalog

● [ 2 ] Trino / DuckDB / Dremio 可以使用各自的 Iceberg connector 读取同一张表。

[ 3 ] 所有引擎都能看到 RisingWave 持续更新的同一份 Iceberg 数据。

● 无需副本，无需专有表格式 — 只有由 RisingWave 写入的标准 Iceberg。

---

从本地笔记本电脑到生产集群：部署选项

● 部署选项：从笔记本电脑到集群

[ 1 ] 本地（用于学习和原型开发）：

● 使用 Docker 运行 RisingWave、Kafka、MinIO 和 Lakekeeper。

● 非常适合在笔记本电脑上试验流连接和 Iceberg 表。

[ 2 ] 生产环境（用于实际工作负载）：

● 通过 Kubernetes + Helm 部署 RisingWave 和技术栈的其余部分。

● 使用适合你的环境的存储类、资源限制和持久化配置。

● 在 RisingWave 中使用相同的 SQL 和模式 只不过更加持久、可扩展且自动化程度更高。

简化传统 Iceberg 技术栈

● 传统 Iceberg 部署通常需要：

● 单独的流处理引擎。

● 独立运行的 Iceberg writer 作业。

● 外部压缩和维护工作流。

● 额外的衔接机制，以确保 catalog、writer 和存储保持一致。

● 使用 RisingWave：

● [ 1 ] 流数据库负责数据摄取、连接、物化视图和 Iceberg 写入。

● [ 2 ] REST catalog + MinIO 让一切保持完全开放且可互操作。

● 更少的活动组件、更低的运维负担。

使用 RisingWave 的参考架构

● 可以将系统视为以 RisingWave 为中心的三个层级：

[ 1 ] 数据流 → RisingWave 表。

● Kafka 主题会成为 RisingWave 中的流式表。

[ 2 ] 表 → RisingWave 物化视图。

● 流连接和聚合会成为实时 MV（logistics_joined_mv、truck_fleet_overview）。

[ 3 ] 视图 → 流式 Iceberg 表。

● RisingWave 只需少量配置和一条 INSERT....SELECT，即可将 MV 转换为流式 Iceberg 表。

● 一旦将 RisingWave 视为“流式 SQL + Iceberg 引擎”，就能在许多领域复用这一模型。

物流以外的可复用模式

● RisingWave + Iceberg 模式适用于：

● 电子商务：订单、库存、定价、客户事件。

● FinTech：交易、余额、风险信号。

● 工业 IoT：机器、传感器、警报、维护。

● 电信：会话、使用量、QoS 指标。

● 只要有多个实时数据流，并且需要开放式长期存储，就可以用同样的方式使用 RisingWave MV 和 Iceberg 表。

要点总结（RisingWave + Iceberg）

● 结合 Kafka、RisingWave、REST catalog、MinIO 和 Iceberg 的参考架构。

● 实用模式：多路流连接、KPI 视图，以及从 RisingWave 原生写入 Iceberg。

● 无需自定义 writer、临时拼凑的压缩作业或严重的供应商锁定，即可获得实时物流分析。