2026-08-28
每日一文 · 长文精读
Article: Beyond Offset Lag: Computing Time in Queue for Apache Hudi Data Lake Pipelines at Petabyte Scale
超越偏移量滞后：在 PB 级规模下为 Apache Hudi 数据湖管道计算排队时间
作者：Srikanth Mamidala · InfoQ 原文
摘要：本文介绍 Twilio 在大规模 Hudi 数据湖管道中如何用“排队时间”替代传统的 Kafka Offset Lag 作为数据新鲜度指标。通过从 S3 中的 Hudi commit 文件提取 checkpoint，定位 Kafka 中尚未消费的最早消息的时间戳，计算其与当前时间的差值。该方法无需改动现有管道，支持自定义 SLA，并与偏移量监控互补。
Kafka Offset lag tells you how far behind a consumer is rather than telling you how old the data is.
Kafka 偏移量延迟告诉你的是消费者落后了多少，而不是数据本身有多旧。
For Apache Hudi pipelines, these measurements are two completely different things.
对于 Apache Hudi 管道而言，这两个度量完全是两回事。
The time-in-queue metric is computed by reading the Kafka checkpoint from the latest Hudi commit file in S3, seeking to that offset in the Kafka topic, and measuring the timestamp delta between that message and the current time.
排队时间指标的计算方式是：从 S3 中最近的 Hudi commit 文件读取 Kafka checkpoint，在 Kafka 主题中定位到该偏移量，然后测量该消息时间戳与当前时间之间的差值。
Changes to producers, consumers, or existing pipeline infrastructure are not required.
无需对生产者、消费者或现有管道基础设施做任何改动。
The algorithm must handle the case where the latest Hudi commit contains no checkpoint metadata.
该算法必须处理最新 Hudi commit 中不包含 checkpoint 元数据的情况。
Once deployed, time-based lag becomes a first-class data contract metric.
一旦部署，基于时间的延迟就成为一等数据契约指标。
Offset monitoring and time-lag monitoring are complementary.
偏移量监控和时延监控是互补的。
At Twilio, the data lake is the foundation for analytics, reporting, and machine learning across the company’s product lines: messaging, email, voice, and more.
在 Twilio，数据湖是公司各产品线（消息、电子邮件、语音等）分析、报表和机器学习的基础。
The pipelines that feed this lake use Apache Hudi Delta Streamer to land data from Kafka, processing over five trillion records monthly as of Q4 2025, across self-hosted Kafka clusters, peaking at 12.9 million messages per second on Cyber Monday 2025.
为数据湖供数的管道使用 Apache Hudi Delta Streamer 将 Kafka 数据落湖，截至 2025 年第四季度每月处理超过 5 万亿条记录，横跨自建 Kafka 集群，在 2025 年网络星期一达到每秒 1290 万条消息的峰值。
We were looking for a signal that was actionable without adding any overhead to live pipelines.
我们一直在寻找一种可操作的信号，而且不会给在线管道增加任何额外开销。
Traditional consumer lag metrics like consumer offset lag (records-lag-max) and even Hudi’s kafkaDelayCount looked fine.
传统的消费者延迟指标，如 consumer offset lag (records-lag-max) 乃至 Hudi 的 kafkaDelayCount，看起来都很正常。
It seemed like consumers were keeping up with Kafka, but downstream analytics teams kept reporting stale data that was sometimes hours old.
看起来消费者跟得上 Kafka 的速度，但下游分析团队不断报告陈旧数据，有时数据已落后几个小时。
The issue wasn't Kafka throughput; it was a visibility gap.
问题不在于 Kafka 吞吐量，而在于可见性缺口。
Hudi Delta Streamer manages its own checkpoints, which are stored alongside the table data in S3 and separate from Kafka's consumer group offset tracking.
Hudi Delta Streamer 管理自己的 checkpoint，这些 checkpoint 与表数据一起存储在 S3 中，并与 Kafka 的消费者组偏移量跟踪分开。
Standard lag monitoring tools like Burrow track a consumer group’s committed offsets, which Hudi doesn’t populate by default, so they had no awareness of whether Hudi had actually committed that data to the lake.
Burrow 等标准延迟监控工具跟踪消费者组已提交的偏移量，而 Hudi 默认不会填充这些偏移量，因此这些工具无法感知 Hudi 是否真的把数据提交到了数据湖。
The real question we needed to answer was how far behind is the latest Hudi commit from the messages currently sitting in the Kafka topic?
我们需要回答的真正问题是：最新的 Hudi commit 与当前 Kafka 主题中的消息相比落后了多少？
How far behind is the Hudi job from messages in the Kafka topic?
Hudi 任务相对于 Kafka 主题中的消息落后了多少？
In other words, we wanted to compute and report how long it has been since the first unconsumed message arrived in the Kafka topic after a successful Hudi commit.
换句话说，我们希望计算并报告：在一次成功的 Hudi commit 之后，第一条未消费消息到达 Kafka 主题已经过去了多久。
HoodieStreamer (formerly HoodieDeltaStreamer) utilizes a checkpoint mechanism to track exactly what data has been ingested and prevent reprocessing of that same data.
HoodieStreamer（原名 HoodieDeltaStreamer）利用 checkpoint 机制精确跟踪已摄入的数据，并防止重复处理相同数据。
For Kafka sources, this checkpoint represents the exact topic offsets (per partition) or timestamps that have been successfully processed and committed to storage.
对于 Kafka 数据源，该 checkpoint 表示已成功处理并提交到存储的精确主题偏移量（按分区）或时间戳。
For checkpoint storage, the checkpoint is embedded directly in the .hoodie commit files as deltastreamer.checkpoint.key.
在 checkpoint 存储方面，checkpoint 以 deltastreamer.checkpoint.key 的形式直接嵌入 .hoodie commit 文件中。
From this timeline of events, we can find the latest commit using the Apache Hudi SDK and extract the per-partition offsets that were last successfully written to the lake.
从这个事件时间线中，我们可以使用 Apache Hudi SDK 找到最新 commit，并提取上次成功写入数据湖的每个分区的偏移量。
What makes this approach practical is that it requires nothing new from the pipelines themselves.
这种方法的实用性在于，它不需要管道本身提供任何新东西。
The metrics reporter is purely an external observer that reads artifacts the system already produces, requiring neither new instrumentation nor producer changes.
metrics reporter 纯粹是一个外部观察者，只读取系统已产生的产物，既不需要新的埋点，也不需要修改生产者。
The metrics reporter runs every fifteen minutes in production.
metrics reporter 在生产环境每十五分钟运行一次。
Walk through commits in reverse chronological order, the most recent first, to find the latest commit that contains a deltastreamer.checkpoint.key.
按时间倒序遍历 commit（最新的在前），找到包含 deltastreamer.checkpoint.key 的最新 commit。
Seek to the checkpoint offset in each Kafka partition.
在每个 Kafka 分区中定位到 checkpoint 偏移量。
Read that message and get its timestamp X.
读取那条消息并获取其时间戳 X。
Compute lag: currentTimestamp - X = how long that data has been waiting
计算延迟：currentTimestamp - X = 该数据已经等待了多长时间
Cap at seven days if lag exceeds the threshold.
如果延迟超过阈值，则上限设为七天。
If no valid checkpoint is found within the search depth, the reporter suppresses the metric entirely rather than publishing a misleading value.
如果在搜索深度内找不到有效 checkpoint，reporter 会完全抑制该指标，而不是发布一个误导性的值。
To clarify this approach, here is a full walkthrough with real numbers.
为了阐明这种方法，下面用一个真实数值做完整演练。
Consider a topic orders-events with three partitions.
假设有一个包含三个分区的主题 orders-events。
The latest Hudi commit contains this checkpoint: orders-events,0:1200,1:980,2:1450
最新的 Hudi commit 包含如下 checkpoint：orders-events,0:1200,1:980,2:1450
These are next-to-read offsets.
这些是待读取的偏移量。
Hudi has committed all messages up through offset 1199 on partition 0, 979 on partition 1, and 1449 on partition 2.
Hudi 已提交到分区 0 的 offset 1199、分区 1 的 979 和分区 2 的 1449 为止的所有消息。
The earliest timestamp across all partitions is forty-five minutes old.
所有分区中最早的时间戳是四十五分钟前。
That is the oldest message currently waiting to be committed to the lake.
那就是当前等待写入数据湖的最旧消息。
If the pipeline’s slaInMinutes is thirty, the SLA ratio is 45/30 = 1.5, capped at 1.0, which is fully breached.
如果管道的 slaInMinutes 为 30，则 SLA 比率为 45/30 = 1.5，封顶为 1.0，即完全违反。
The ratio is always capped at 1.0 because a pipeline cannot be more than fully breached.
该比率始终封顶为 1.0，因为管道不可能比“完全违反”更严重。
阅读理解
1. What does the 'time-in-queue' metric measure?
2. Why do standard consumer lag metrics like records-lag-max fail to reflect real data freshness in Hudi pipelines?
3. What does the metrics reporter do when the latest Hudi commit contains no checkpoint metadata?