当前位置：首页 > news >正文

Flink Traces 用 Span 把“到底慢在哪”讲清楚

news 2026/3/26 19:04:15

1. Flink Traces 的核心模型：Trace = 一棵 Span 树

Span表示一次发生在某段时间内的过程（有开始、结束、属性）。
Flink 当前支持的 trace 结构是：一棵树（tree of spans）。
重要限制：所有子 span 必须在一次addSpan调用里“一次性”上报，不能先报 parent，过一会儿再补 child，也不能单独上报 child/parent。

这意味着：如果你要记录一个过程的多级细分（父 span + 多个子 span），要在内存里构造好整棵树后再提交。

2. 在用户函数里上报 Trace：从 RuntimeContext 拿到 MetricGroup

Flink 把 tracing 的入口放在 MetricGroup 上：

你需要在继承RichFunction的用户函数中，通过：
getRuntimeContext().getMetricGroup()
拿到MetricGroup，然后调用：
MetricGroup#addSpan(SpanBuilder)

2.1 Java 示例：上报一个父 Span + 多个子 Span

下面的写法体现了两个关键点：
1）父 span 可以设置 start/end（可选）
2）子 span 必须在同一次 addSpan 中一起提交（包括多级 children）

publicclassMyClass{privateMetricGroupmetricGroup;voiddoSomething(){longstartTs=System.currentTimeMillis();// ... 业务逻辑longendTs=System.currentTimeMillis();metricGroup.addSpan(Span.builder(MyClass.class,"SomeAction").setStartTsMillis(startTs)// Optional.setEndTsMillis(endTs)// Optional.setAttribute("foo","bar")// Optional.addChild(Span.builder(MyClass.class,"ChildAction")// Optional).addChildren(List.of(Span.builder(MyClass.class,"AnotherChildAction"))));}}

你可以把它理解成：“一次 addSpan = 提交一棵 span 树”。

2.2 Python（PyFlink）怎么写？

你提供的内容里 Python 部分还没展开具体 API。思路上是一样的：

在 RichFunction 生命周期里拿到 runtime context
从 metric group 构造 span builder
一次性提交包含 children 的 span 树

如果你用的是 PyFlink，建议你优先确认你当前 Flink 版本对应的 PyFlink tracing API 是否已对齐（有些能力可能先在 Java 侧完善）。

3. TraceReporter：把 trace 发到外部系统

Span 上报之后，最终会交给TraceReporter输出到外部系统（比如你现有的可观测平台）。

你给的原文里提示：TraceReporter 的配置需要参考 “trace reporters documentation”。这里我不硬编配置项（不同版本/发行版可能不同），但给你一个落地建议：

先把 TraceReporter 打到日志/本地（最容易验证是否“有数据”）
再接入链路系统（如 OpenTelemetry 体系、Jaeger/Tempo/自研平台等）

验证路径很简单：先确认 Flink 侧确实在 addSpan 后产生输出，再做下一步接入。

4. Flink 内置 System Traces：Checkpoint 与 Job Initialization

除了你自己上报的 trace，Flink 也会自动上报一些系统级 traces。你提供的内容里重点是两类：

Checkpointing
Job Initialization（作业初始化/恢复）

并且它们的特点是：
Flink 会在事件到达终态（COMPLETED 或 FAILED）时，上报一个“单 span trace”，覆盖整个事件生命周期。

4.1 Checkpoint Trace（CheckpointStatsTracker）

Scope：org.apache.flink.runtime.checkpoint.CheckpointStatsTracker
Name：Checkpoint

包含的关键 Attributes（你在排障时非常有用）：

startTs：checkpoint 开始时间戳
endTs：checkpoint 结束时间戳
checkpointId：checkpoint ID
checkpointedSize：本次实际 checkpoint 的状态大小（bytes），增量场景可能小于 fullSize
fullSize：引用的完整状态大小（bytes）
checkpointStatus：FAILED/COMPLETED
checkpointType：例如"Checkpoint","Full Checkpoint","Terminate Savepoint"等
isUnaligned：是否为 unaligned checkpoint

怎么用它做判断：

checkpointedSize很小但耗时很长：可能不是“上传体积”导致，而是对齐、反压、阻塞等
isUnaligned=true且后续恢复慢：要重点关注初始化 trace 里的 buffer restore 相关阶段

4.2 JobInitialization Trace

Name：JobInitialization

Attributes 里把恢复/初始化拆得很细，尤其适合定位“恢复为什么慢”：

基础字段：

startTs/endTs
checkpointId（可选）：从哪个 checkpoint 恢复
fullSize：用于恢复的 referenced state 完整大小

聚合耗时字段（Flink 会对所有 subtasks 做 max/sum 聚合）：

(Max/Sum)MailboxStartDurationMs：subtask 创建到类/对象初始化完成的时间
(Max/Sum)ReadOutputDataDurationMs：读取 unaligned checkpoint 输出 buffers 的时间
(Max/Sum)InitializeStateDurationMs：初始化 state backend 的时间（含下载 state 文件）
(Max/Sum)GateRestoreDurationMs：读取 unaligned checkpoint 输入 buffers 的时间

RocksDB Incremental（可选，当前只提到对 RocksDB 增量恢复支持）：

(Max/Sum)DownloadStateDurationMs：从 DFS 下载 state 文件耗时
(Max/Sum)RestoreStateDurationMs：本地化完成后的 restore 耗时
(Max/Sum)RestoredStateSizeBytes.[location]：按位置统计恢复的状态大小（LOCAL_MEMORY/LOCAL_DISK/REMOTE/UNKNOWN）
(Max/Sum)RestoreAsyncCompactionDurationMs：增量恢复后的异步 compaction 耗时

怎么用它快速定位瓶颈：

DownloadStateDurationMs很高：网络/DFS 吞吐或并发连接数是重点
InitializeStateDurationMs很高但下载不高：更多是 state backend 初始化与加载开销（例如 RocksDB 打开、恢复元数据）
GateRestoreDurationMs / ReadOutputDataDurationMs很高：unaligned checkpoint 的 buffer restore 成本要重点评估（尤其高并发/高吞吐作业）