构建流式 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、临时拼凑的压缩作业或严重的供应商锁定，即可获得实时物流分析。