Pinterest 把分区完成度写进 Iceberg 快照摘要，用不回退的水位线换掉外部协调；Kafka 用 KIP-1279 把跨集群复制收进 broker，offset 与 epoch 逐字节对齐；Apache Fluss 1.0.0 带来 Rust Gateway 与四语言客户端；Uber 在 800 万 QPS 的推理路径上直接记下真实特征

本期四条分属湖仓表格式、消息中间件、流存储与机器学习数据管道，动作却落在同一处：把一份原本放在系统外面的账，搬进离数据最近的那一层。Pinterest 把「这个小时的数据到齐了没」从外部协调服务搬进 Iceberg 的快照摘要与表属性；Kafka 用 KIP-1279 把「复制到哪个 offset 了」和 offset 映射表从 MirrorMaker 2 的 Connect 集群搬进 broker 自己的 KRaft 元数据与一个内部 topic；Apache Fluss 1.0.0 把「没有 JVM 的 Agent 怎么读写流存储」搬进一个用 Rust 写的无状态 Gateway，并让 Spark 用一条 SQL 拼起热层与湖层；Uber 把「模型训练时到底该用哪个特征值」从离线重算搬成推理端点上的实时记录。四处搬运省掉的都是一个外部组件——协调服务、Connect 集群、SDK 与冷热合并逻辑、离线特征 join——代价是新账本的表达力有限，这些限制条款才是本期最值得先读的部分。

Pinterest：把「到齐了没」写成 Iceberg 上的一个不回退水位线

事件概述

9 月 25 日，Pinterest 工程团队的 Qianrui Zhang、Kanchi Masalia、Liang Mou、Yi Pan 在 Pinterest Engineering Blog 发表《Partition Finalization in Pinterest’s Next-Generation DB Ingestion Framework》。这是其新一代数据库入湖框架系列的第三篇：第一篇讲基于 Kafka、Flink、Spark 与 Iceberg 的入湖框架，第二篇讲自动 schema 演进，这一篇解决的是一个听起来很朴素的问题——下游什么时候可以读。

在旧的批处理体系里，这个问题不需要回答。一批作业把源库整读一遍、一次性写完，作业结束即分区完成，下游可以直接开始。改成流式之后，Flink 作业持续把变更事件提交成一个个小批次；一条记录的事件时间（变更在源库发生的时刻）可以晚于它被处理的时刻，于是分区在整点之后仍会继续收到数据，「分区里有数据」不再意味着「分区的数据齐了」。Pinterest 需要在它支持的数千条流水线上统一地推断出这一点，而不是让每条流水线各自约定一种标记方式。

技术解析

• 从记录里取出事件时间：每条流水线配置哪一列承载事件时间，一个可插拔的 extractor 读取该列并归一化为时间戳。换一张表只需把 extractor 指向那张表的事件时间列。
• 事实收集：两次 checkpoint 之间，writer subtask 内的 accumulator 聚合本窗口内所有记录的事件时间，核心是最小值与最大值，并额外保留 95、99 分位数，以保留分布形状而不只是极值。
• 事实压进快照摘要：checkpoint 提交时，这些统计写进 Iceberg 快照的 summary——Iceberg 原本就为每次提交维护的元数据。分位数用 t-digest 存，摘要大小恒定在几百字节量级，无论这个 checkpoint 处理了一千条还是几百万条；t-digest 可合并，各 subtask 的局部结果在 committer 处合成一份。每次提交自带一份自描述记录，不需要额外的外部存储去保持同步。
• 从事实到判断：每次提交后，框架对近期若干次提交的统计跑一个算法，产出水位线。默认取最近 X 次提交各自最小事件时间的最小值，再向下取整到小时边界。
• 水位线不回退：迟到数据若会把水位线往回拉，取值保持不变，已完成的小时仍是完成的。此时应用会向数据消费方发一条告警，标记「有迟到数据在水位线推进之后到达」。
• 事实与判断分开存放：事实（每次提交的统计）落在各自的快照上，判断（分区完成水位线）是表级单值，存为 Iceberg 表属性。下游用一次普通元数据查询读到它，不需要旁路通道或独立服务。
• 跟着数据一起走：完成度在 CDC 表上算出来，但很多下游读的是 Spark upsert 产出的 base 表，那张表有数据没有信号。upsert 作业每次运行时从 CDC 表读出水位线，在同一次提交里写进 base 表——信号顺着数据走到下游真正读的地方，传播动作只是复制一次元数据。
• 实现落在 sink 的两处扩展上：accumulator 在每个 writer subtask 内累积、checkpoint 时快照并重置，committer 合并各 subtask 结果（合并必须与顺序无关）；commit processor 在 Iceberg 提交前后各挂一个钩子，提交前可设置快照属性或直接抛异常让 Flink 重试该 checkpoint，提交后读刚写入的统计、跑水位线算法、更新表属性。整个过程在提交周期内完成，不依赖任何外部服务。

行业影响

流式入湖用分钟级新鲜度换掉了批处理免费附送的完整性保证，这条账过去由各团队自己用「等一个 marker 文件」「等 N 分钟」这类约定来还。Pinterest 的做法是把完成度变成表上一个可读的属性，于是在新鲜度与完整性之间取舍成了一个旋钮，而不用每个下游各自猜。它对生态的意义还在于位置选得克制：没有新增存储、没有新增服务，只用了 Iceberg 已经存在的两个位置——快照 summary 放事实、表属性放判断——任何能读 Iceberg 元数据的引擎都能消费这个信号，传感器不必换实现。

限制同样写在文里：水位线是保守估计，取的是最近 X 次提交最小值的下界并向下对齐到小时，晚到的数据只能靠告警通知，已完成的小时不会重开；X 的具体取值未给出，算法参数属于「策略旋钮」而非固定语义。把判断写进表属性也意味着它随表走、不随查询走——某次查询想用更激进的完成度标准，仍得自己另算。

相关链接

• Partition Finalization in Pinterest’s Next-Generation DB Ingestion Framework
• Data Engineering Weekly #289（本期多条线索的出处）

---

Apache Kafka：KIP-1279 把跨集群复制收进 broker，offset 逐字节对齐

事件概述

9 月 22 日（页面最后更新于 9 月 25 日），Red Hat 开发者网站刊文介绍 Apache Kafka 通过 KIP-1279 引入的原生集群镜像（Cluster Mirroring）。它的做法是把跨集群复制直接嵌进 broker：目标端 broker 用 follower 已经在用的 fetch 协议从源集群拉取已提交记录，逐字节追加到本地分区日志，保留 offset、压缩格式与 consumer group 状态。故障转移被压缩成两步——停止镜像，把客户端指向新集群。

MirrorMaker 2（MM2）自 Kafka 2.4 起是这件事的标准答案。KIP-1279 与它的对比，官方给了一张很直接的对照表：

维度	MirrorMaker 2	Cluster Mirroring
部署形态	外部 Connect workers	嵌入 broker
压缩	解压后重新压缩	逐字节透传
Offset	经 topic 做有损转换	跨集群完全一致
Consumer 故障转移	查询 offset 同步 topic	直接接管，无需转换
Unclean election	无处理	完整日志收敛
源兼容性	Kafka 2.0+	Kafka 2.1+
监控	Connect 专用工具链	标准 broker JMX 指标

技术解析

• 三个组件：MirrorMetadataManager 是编排者，实现 MetadataPublisher、对 KRaft 元数据日志的变化做出反应，默认每 60 秒刷新一次源端元数据，负责发现新 topic、同步 topic 配置、拉取 consumer group offset，并校验源集群 ID 未变（防止有人把镜像误指向另一个集群而静默产生数据损坏）；ClusterMirrorCoordinator 负责状态持久化，管理内部 compacted topic __mirror_state，默认 50 分区、副本因子 3，用 leader epoch 与 state epoch 做乐观并发控制；MirrorFetcherThread 扩展 Kafka 的 AbstractFetcherThread，与集群内复制共用同一基类，每个线程持有独立的 NetworkClient 与 per-mirror 认证凭据，使 SASL/SSL context 在不同镜像之间隔离。
• Epoch fencing：broker 向 controller 发 BumpLeaderEpochs 请求，local leader epoch 每次 +10，re-bump 阈值 3，目标是保证目标端 leader epoch 不小于源端，否则目标端 consumer 会因源端已提交 epoch 超过本地 epoch 而拒绝 leader。
• 两阶段截断：状态机为 LOG_ALIGNMENT → EPOCH_FENCING → MIRRORING，另可从 MIRRORING 进入 ULE_RECOVERY、PAUSED、STOPPED，任一状态出错转 FAILED。首次镜像或源端不支持时截断到零从头镜像；否则只按 LME（Last Mirror Epoch）记录的上次镜像 offset 与 epoch 做增量截断。若源端支持 Fetch v12+，fetch 响应里的 diverging epoch 信息允许内联截断到精确分歧点，不支持则回退到显式 OffsetsForLeaderEpoch 请求。
• 停止序列是确定性的：移除 fetcher 线程 → 记录 LME 供将来 failback → bump 本地 epoch → 为所有 in-flight 事务追加 ABORT marker，使 READ_COMMITTED consumer 不会看到悬挂事务 → 写入 MIRROR_PID_RESET control record 让 ProducerStateManager 中所有 producer state 过期，新 producer 拿到无冲突的新 PID → 分区在目标端变为可写。回切时反向镜像会查到故障转移期间记录的 LME，执行增量截断而不是重复制整个数据集。
• 版本边界：源端最低支持到 Kafka 2.1，依赖 Kafka 4.0 引入的 client/broker 前向兼容；目标端必须是 KRaft。传统迁移路径需要按 2.x → 3.x → 4.x 逐级升级并原地做 ZooKeeper 到 KRaft 的元数据迁移，KIP-1279 允许直接从 2.1 的 ZK 集群镜像到新的 KRaft 集群，旧集群可以一直跑到准备下线。
• 未覆盖的部分：复制是异步的，RPO 取决于故障发生时的复制延迟，非计划故障转移会丢掉已生产但未复制的记录；零 RPO 要等 KIP-1360 的同步模式，该 KIP 仍在计划中。复制本身用 READ_UNCOMMITTED 隔离级别，未提交事务记录会先于源端最终决定到达目标端（这是刻意的取舍，等事务完成会增加延迟与复杂度）。源集群 ID 变更、源 topic 被删除属于不可重试错误，需要人工介入。源端发生 unclean leader election 时，默认配置 mirror.unclean.leader.election.enable=false 会让分区进入 FAILED 直接停止镜像；置为 true 则进入 ULE_RECOVERY，等待所有 assigned replica（不只是 ISR 成员）收敛后恢复，并接受源端 ULE 带来数据丢失。同级路线上的 tiered storage 集成与 diskless topics 支持均属未来工作。

行业影响

MM2 的问题从来不是不能复制，而是复制这件事被放在了 Kafka 外面：一套要独立 provision、监控、扩容的 Connect 集群，加一张用来做偏移翻译的同步 topic，而翻译本身是有损的。KIP-1279 删掉的正是这两样外部件，以及翻译带来的语义损失——offset 完全相同，连 topic compaction 留下的空洞都一致，consumer group 在源端提交的 offset 就是目标端的 offset，故障转移后不重放也不丢位。它顺带打通的另一条路更实用：从 ZooKeeper 时代 Kafka 2.1 的旧集群直接镜像到新的 KRaft 集群，不必逐级升级、不必原地迁移元数据。对一个升级窗口常常按季度排期的行业来说，这条路径的价值未必小于灾备本身。

需要读清楚的代价是异步复制下的 RPO，以及目标端必须是 KRaft 这条硬约束。把复制收进 broker 也意味着复制流量与正常流量争抢同一个进程的资源，文档给出的控制手段是目标端可配置的复制速率限制与源端沿用现有 client quota 机制——节流旋钮在，但隔离性不如外部集群。

相关链接

• Data liberation: Apache Kafka’s native cluster mirroring（Red Hat Developer）
• KIP-1279: Cluster Mirroring

---

Apache Fluss 1.0.0：毕业后的首个大版本，Gateway 用 Rust 写，四语言客户端进主干

事件概述

9 月 21 日，Apache Fluss PMC 成员 Keith Lee 与 PMC Chair Jark Wu（伍翀）联合发布《Announcing Apache Fluss 1.0: Real-Time Data Foundation for AI》，GitHub 上的 v1.0.0 tag 于 9 月 22 日打出。这是 Fluss 从 Apache 孵化器毕业成为顶级项目之后的第一个大版本，累计超过一千次提交、约 45 万行新增代码，是此前数个版本新增量的三倍左右。Fluss 源自阿里云实时数据湖团队，2024 年在 Flink Forward Asia 上开源，定位是「为实时分析构建的流存储」，主张把 Kafka 的实时性与湖表的可分析性放进同一个存储层。

技术解析

• Fluss Gateway：一个用 Rust 编写、复用 Fluss Rust 客户端内核的无状态 REST 服务，把 REST 请求翻译成原生客户端操作。不需要 JVM，可独立部署与扩容，多实例可在负载均衡后并列服务。它支持集群发现与库/表/分区的元数据 API、库表分区 DDL、带 schema 感知的 JSON 解码，以及批量 append、upsert、partial update、delete（逐行返回结果）。1.0 中为 preview：记录读取与 HTTP 调用方认证都不包含在内。路线图上还有 MCP、PostgreSQL 线协议与 gRPC，目标是把它做成应用、数据工具与 Agent 共用的一层接入面。
• Gateway 也是入湖通道：目标表开启 Lakehouse 后，独立的 tiering service 会把经 Gateway 写入的记录在几分钟内搬进湖仓，把「HTTP 写入」与「分钟级湖仓新鲜度」接在一起。
• 四语言客户端进主干：Rust、Python、C++ 客户端首次进入主仓库并与主项目一同发布，与 Java 客户端并列。三者共享一个 Rust 内核，承担连接管理、协议通信、元数据访问与读写；Python 与 C++ 通过语言绑定暴露能力。Python 侧的两个典型用法是 new_lookup().create_lookuper() 查 Flink 持续更新的 user profile 与特征值，以及 new_append().create_writer() 直接写 PyArrow RecordBatch，免去逐行转换。
• 湖仓侧：新增 Hudi 支持，并补上 Paimon 历史分区的 lookup 与更新，让 Agent 取历史数据更快。
• 引擎侧：Flink 内置 bitmap 函数与 Fluss 的 RoaringBitmap 聚合表打通；Spark 新增 batch Union Read——对开启 table.datalake.enabled 的表自动合并 Fluss 热层最新数据与湖层历史数据，一条 SQL 拿到全量视图；另新增时间窗增量读 fluss_incremental_between_timestamp('...', start, end)，窗口为半开区间 [start, end)，日志表返回窗口内追加的记录，主键表返回窗口内发生过变更的每个键在窗口结束时的最新有效值，delete 事件不返回。
• 生产运维：Coordinator 高可用（多 Coordinator 部署，一主多备，leader 故障后选举接管；滚动升级时可逐台重启）；TabletServer 支持多个本地数据目录与多个远端数据目录；磁盘使用率写保护与可配置恢复阈值，磁盘写被阻塞时阻止 KV leader 晋升；基于 RocksDB Level-0 文件压力的协作式 KV 写反压；table.kv.ttl 提供主键表行级 TTL，table.log.local-ttl 提供独立的本地日志保留。另新增腾讯云 COS 文件系统、InfluxDB metrics reporter。
• 升级代价：从 0.9 升级前必须先读 1.0 升级说明，官方点名三处不兼容——Lakehouse 升级顺序、ACL 变更、AWS 凭证 provider 迁移。

行业影响

Fluss 1.0 把「流存储」这条主张往外扩了两个方向。横向上，Gateway 让没有 JVM 的 Agent 与 Web 服务用 HTTP 就能发现元数据、建表、写入，把入口从「必须装对应语言的 SDK」放宽到「会发 HTTP 就行」；纵向上，Spark 的 Union Read 与时间窗增量读把「热层 + 湖层」的拼接、以及按提交时间抽取增量，变成了引擎侧的一条 SQL，不必在自己的管道里维护冷热合并逻辑。它和本期其余三条落在同一处：入湖进度、分区完成度、复制偏移这些原本要外部系统记的账，正在被搬进存储与表格式自己。

边界要一并读。Gateway 现在是 preview 且缺读路径与认证，这意味着它当下只能当「写入与元数据入口」用，把它当通用读接口会撞空；MCP 与 PostgreSQL 线协议属于路线图而非已交付能力；增量读不返回 delete 事件，用它做增量抽取的表需要有别的方式感知删除。从 0.9 升级的三处不兼容变更，对已经在生产上跑了 Lakehouse 的集群尤其需要先演练。

相关链接

• Announcing Apache Fluss 1.0: Real-Time Data Foundation for AI
• Apache Fluss Downloads
• apache/fluss v1.0.0 release notes（GitHub）
• Fluss 1.0 Upgrade Notes

---

Uber：在推理端点上记下模型真正用到的特征，替换离线重算

事件概述

9 月 22 日，Uber 的 Paarth Chothani 在 Uber Engineering Blog 发表《Taming the ML Firehose: Scaling Feature Consistency》，讲的是 Uber Eats 推荐模型上的一个老问题：模型在生产上看到的值，与它被训练时的值不是同一份。

文章给了两个具体的失配。训练时模型学的是固定词表，语言编码为 en-US、fr-FR；到了线上推理，同样的特征传来的是 en。离线侧的另一个管道又用下划线代替连字符，于是训练数据里出现 jp_JA、线上是 jp-JA。这类不一致不抛错、不掉线，只是让强信号静默退化，特征分布漂移而预测稳定性变差，排查起来要花数周。

技术解析

• 做法：在推理端点上记录模型真实消费到的那组特征值，把这些记录作为下一轮训练的权威来源。日志发生在打分那一刻，记录的是模型实际看到的值，调用方改动很小。
• Feature Allow List：预测请求往往携带许多模型并不使用的特征。推理端点在返回预测前先查一张允许名单，只把批准的特征序列化发到 Kafka，payload 缩小 4 到 5 倍。
• 链路：推理服务计算特征 → 按允许名单序列化写入 Kafka → Eats 应用端独立上报 impression 事件 → Flink 把 impression 与先前记录的特征 join，拼出「打分时的真实特征 + 用户后续行为」的完整记录 → 写入另一个 Kafka topic → 落离线存储。每条记录已自带在线特征值，训练时不必再做昂贵的离线特征 join。
• 为什么必须选择性记录：推荐系统一次请求对数千个候选打分，经过滤与排序后只剩约 40%，最终真正成为曝光的只有 5%。全量记录会把绝大多数从未出现在用户面前的候选都存下来，对训练几乎不增加价值。
• 规模约束：Transformer 推荐系统的打分流量在 800 万 QPS 量级。全量记录换算下来是每天数千亿条记录、每月约 5 万亿行、约 1.7 PB/天；算上网络、Kafka 与存储的端到端成本，会推到不可行的量级。
• 结果：关键特征的不一致率从原先超过 10% 降到 0%；多个优先级特征的 SLA 从多天延迟缩短到小时级。
• 验收方式：积累足够数据并完成校验后，用新来源训练模型；同一时间范围再用旧 ETL 管道产出的表训练一份做对照，在 AUC、MAP 等指标持平或更好时才进入下一轮实验。

行业影响

训练与推理的特征不一致通常被当成工程卫生问题，靠规范、靠 review、靠事后对账来治。Uber 的做法换了个方向：不去消除两条路径的差异，而是让训练直接采用推理路径上真实发生的记录，差异被绕过而不是被修复。这与本期其余三条是同一个动作——把「当时到底发生了什么」的账，交给当时离数据最近的那一层来记，而不是事后由另一套系统重建。绕过之后省掉的不只是偏差，还有离线特征 join 的计算与那批互相依赖的 ETL 作业。

代价是日志本身成了新的成本中心，且规模约束相当硬：800 万 QPS 下全量记录不可行，于是必须有允许名单与选择性记录，而「哪些特征值得记」「哪些候选值得记」又变成一组需要维护的策略。文中未给出新的基础设施成本数字，也未说明允许名单由谁维护、如何随模型迭代更新——这两处是这套方案能不能长期跑下去的关键，文章没有展开。

相关链接

• Taming the ML Firehose: Scaling Feature Consistency（Uber Engineering Blog）

---

编者评

把四条并起来看，本期反复出现的动作不是加速，也不是扩容，而是搬家：一个原本由外部系统代管的记录，被搬到了离数据最近的那一层。Pinterest 搬的是完成度——从约定式的 marker 与外部协调服务，搬进 Iceberg 的快照摘要与表属性；Kafka 搬的是复制进度与 offset 映射——从 MM2 的 Connect 集群与同步 topic，搬进 broker 自己的 KRaft 元数据与 __mirror_state；Fluss 搬的是冷热拼接与增量抽取——从各家的管道逻辑，搬进 Spark connector 的一条 SQL；Uber 搬的是特征值——从离线重算，搬成推理端点上的实时记录。搬一次就删掉一个组件：一个协调服务、一套 Connect 集群、一段冷热合并代码、一批离线特征 join。

四处搬家能一并发生，是因为承载方的位置本身变宽了。Iceberg 的快照 summary 与表属性本来就在，只是过去没人往里放判断；Kafka 的 KRaft 元数据日志本来就在，只是过去复制状态不归它管；Spark connector 本来就在，只是过去 Union Read 要自己写。换句话说，这不是新造了一层，而是把已有的元数据结构用到了它原本没打算承担的职责上。

值得留意的是每一处都留了口子，且这些口子性质相同：新账本能记的东西比旧账本少。Pinterest 的水位线是保守下界，已完成的小时不会因迟到数据重开；KIP-1279 是异步复制，非计划故障转移会丢未复制的记录，零 RPO 要等 KIP-1360；Fluss 的 Gateway 缺读路径与认证，增量读不返回 delete；Uber 的选择性记录只覆盖 5% 真正曝光的候选，剩下 95% 从未进入训练视野。搬家的收益是删掉一个组件，成本是放弃一部分表达力——判断要不要跟，看的是丢掉的这部分表达力在自己的场景里到底有多常用，而不是看删掉了多少行运维代码。