java

8.1 指标与延迟分析

Flink 作业指标系统与延迟分析：内置 Metrics 类型（Counter/Gauge/ Histogram/Meter）、LatencyMarker 端到端延迟追踪、MetricsReporter 配置、Web UI 指标查看，以及 eventd v3 的自定义指标（QPS、延迟分位数、错误率）实现。

Leeting Yan 2026-11-01 5 分钟阅读 2189 字

为什么需要指标

流处理作业一旦上线，「看不见」意味着「不可控」。你需要知道：

每秒处理多少条数据（吞吐）
从 Source 到 Sink 花了多少时间（延迟）
哪个算子卡住了（反压）
Checkpoint 花了多久（健康度）

Flink 内置了丰富的指标系统，无需额外引入 APM 工具即可获取核心数据。

Flink 指标类型

Flink 支持四种指标类型：

类型	语义	示例
Counter	累加计数	处理记录数、Checkpoint 次数
Gauge	当前值	队列长度、内存占用
Histogram	分布	延迟分布、批处理大小
Meter	速率	每秒处理量（Throughput）

Counter 示例

// RichMapFunction 已经实现了 MapFunction，只需 extends；
// 写成 implements ... , RichMapFunction 是语法错误（类是抽象类，不能放 implements 列表）
public class MyMap extends RichMapFunction<Event, Event> {
    private Counter recordCounter;

    @Override
    public void open(OpenContext ctx) {
        recordCounter = getRuntimeContext()
            .getMetricGroup()
            .counter("recordsProcessed");
    }

    @Override
    public Event map(Event event) {
        recordCounter.inc();
        return event;
    }
}

Gauge 示例

public class QueueMonitor extends RichMapFunction<Event, Event> {
    @Override
    public void open(OpenContext ctx) {
        final AtomicInteger queueSize = new AtomicInteger(0);

        getRuntimeContext()
            .getMetricGroup()
            .gauge("queueSize", () -> queueSize.get());
    }
}

Histogram 示例

@Override
public void open(OpenContext ctx) {
    histogram = getRuntimeContext()
        .getMetricGroup()
        .histogram("eventLatency", new DropwizardHistogramWrapper(
            new com.codahale.metrics.Histogram(new SlidingWindowReservoir(500))
        ));
}

DropwizardHistogramWrapper 来自 org.apache.flink:flink-metrics-dropwizard，它再依赖 io.dropwizard.metrics:metrics-core；两个都不在 Flink 的默认编译依赖里，需要自己加。

LatencyMarker：端到端延迟

Flink 内置的 LatencyMarker 机制，无需修改业务代码即可追踪数据从 Source 到 Sink 的延迟：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 每 10ms 插入一个 LatencyMarker
// 注意不是 setAutoWatermarkInterval：那个控制周期 Watermark，与延迟追踪无关
env.getConfig().setLatencyTrackingInterval(10);

原理

Source → LatencyMarker → Map → KeyBy → Window → Sink
     ↑                                         ↓
   注入时间戳                              到达后计算差值
     → latency = sinkReceiveTime - sourceEmitTime

每个 Source 定期产生特殊的时间戳标记（不含实际数据），沿途经过所有算子时只传递时间戳。Sink 端收到后计算时间差，记录在指标中。

查看延迟指标

Flink Web UI → Metrics → latency 维度：

指标	含义
`latency.source_id.<X>.operator_id.<Y>.operator_subtask_index.<Z>.latency_p50`	从 Source X 到算子 Y 的 P50 延迟
同上，后缀换 `latency_p99` / `latency_mean` / `latency_max` / `latency_min`	其他统计量

指标名由「延迟组 + 统计量后缀」拼成，组里还会按 source_id / operator_id / operator_subtask_index 逐层细分；不要写成 latency.min 这种形式。

注意：LatencyMarker 的频率由配置键 metrics.latency.interval 控制，默认 0（关闭）。pipeline.latency-marker-interval 不是 Flink 1.20 的配置项。

启用：

env.getConfig().setLatencyTrackingInterval(5000);  // 每 5 秒追踪一次

自定义延迟指标

eventd v3 的延迟指标更细粒度——区分业务延迟和系统延迟：

public class LatencyTrackingMap implements RichMapFunction<Event, Event> {
    private transient Histogram processLatency;
    private transient Meter throughput;

    @Override
    public void open(OpenContext ctx) {
        MetricGroup metricGroup = getRuntimeContext().getMetricGroup();
        processLatency = metricGroup
            .histogram("processLatencyMs",
                new DropwizardHistogramWrapper(
                    new com.codahale.metrics.Histogram(
                        new SlidingWindowReservoir(1000))));
        throughput = metricGroup.meter("throughputPerSecond");
    }

    @Override
    public Event map(Event event) {
        long now = System.currentTimeMillis();
        long latency = now - event.getEventTime();
        processLatency.update(latency);
        throughput.markEvent();
        return event;
    }
}

指标	计算方式	健康阈值
`processLatencyMs`	`now - event.getEventTime()`	< 5s
`throughputPerSecond`	Meter 自动计算	与预期 QPS 匹配
`numRecordsIn` / `numRecordsOut`	Flink 内置算子计数器	与上下游速率匹配

想统计「错误率」要额外注意作用域：getRuntimeContext().getMetricGroup() 是子任务级的，Counter("errors") 统计的是这个子任务累计的条数，不带 key、也不会自动算出比率。按业务维度算错误率得自己用 ValueState 按 key 维护成功/失败计数（做法见 10.1 实时指标计算），把结果写进 Redis 等外部存储；MetricGroup 只适合输出子任务级的汇总值。

MetricsReporter 配置

Flink 指标可以通过 Reporter 导出到外部系统。Flink 1.20 的配置键是 metrics.reporter.<name>.factory.class，值是 Reporter 的工厂类，不是 Reporter 实现类本身；旧写法 metrics.reporter.<name>.class 只在兼容路径上还认，新配置不要再用。

Prometheus Reporter

# conf/flink-conf.yaml
metrics.reporters: prom
metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
metrics.reporter.prom.port: 9249

启动后访问 http://localhost:9249/metrics 查看 Prometheus 格式指标。

InfluxDB Reporter

metrics.reporters: influx
metrics.reporter.influxdb.factory.class: org.apache.flink.metrics.influxdb.InfluxdbReporterFactory
metrics.reporter.influxdb.host: localhost
metrics.reporter.influxdb.port: 8086
metrics.reporter.influxdb.db: flink_metrics

JMX Reporter（最简单）

metrics.reporters: jmx
metrics.reporter.jmx.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
metrics.reporter.jmx.port: 9020

用 JConsole 或 VisualVM 连接 localhost:9020 查看。

通过 Web UI 查看指标

本地运行 Flink 作业时，Web UI http://localhost:8081 提供算子级指标：

Job → Job Graph → 点击算子 → Metrics 标签
     └→ records-in                   (Counter)
     └→ records-out                  (Counter)
     └→ latency.source_id.xxx        (LatencyMarker)
     └→ checkpointDuration           (Gauge，如可见)

选择指标后，可以绘制时序图，观察趋势。

告警阈值配置

eventd v2 告警规则（基于指标阈值）。第 8 章仍使用 4.1 的 POJO Event，没有 status 字段；这里约定 type 为 error 的是失败事件：

public class AlertingMap implements RichMapFunction<Event, Event> {
    private transient ValueState<Long> errorCount;
    private transient ValueState<Long> totalCount;
    private transient ValueState<Long> lastAlertTime;

    @Override
    public void open(OpenContext ctx) throws Exception {
        errorCount = getRuntimeContext().getState(
            new ValueStateDescriptor<>("errors", Long.class));
        totalCount = getRuntimeContext().getState(
            new ValueStateDescriptor<>("total", Long.class));
        lastAlertTime = getRuntimeContext().getState(
            new ValueStateDescriptor<>("lastAlert", Long.class));
    }

    @Override
    public Event map(Event event) throws Exception {
        long total = (totalCount.value() == null ? 0 : totalCount.value()) + 1;
        long errors = errorCount.value() == null ? 0 : errorCount.value();
        if ("error".equals(event.type)) {
            errors++;
        }
        totalCount.update(total);
        errorCount.update(errors);

        // 每 60 秒按本 key 的窗口计数检查一次，检查后清零
        long now = System.currentTimeMillis();
        Long last = lastAlertTime.value();
        if (last == null || now - last >= 60_000) {
            long errorRate = errors * 100 / Math.max(total, 1);
            if (errorRate > 5) {  // 5% 错误率
                // 告警是副作用，不是 Event 的字段。
                // 实际项目在这里调用告警客户端，例如：
                // alerts.publish("ERROR_RATE_HIGH", errorRate + "%", now);
            }
            lastAlertTime.update(now);
            errorCount.clear();
            totalCount.clear();
        }
        return event;
    }
}

AlertingMap 里用了 ValueState，必须接在 keyBy 之后（stream.keyBy(Event::userId).map(new AlertingMap())）。直接接在非 keyed 流上，open() 里的 getState 会在运行期抛异常。

不要换成 MetricGroup.counter()：Counter 是子任务级共享、只增不减的，按 userId keyBy 之后读到的仍是「本子任务所有用户的累计值」，而不是当前用户的当前分钟值。上面的 ValueState 按 key 隔离，并在每次检查后 clear()，才与「每用户每分钟错误率」的语义一致。

第 8 章使用 4.1 的 POJO，失败判定只能落在 type 等已有字段上；status() 是 9.1 引入的 v3 record 访问器。告警本身走外部客户端或侧输出，不要为了塞一个告警字段去改共享事件模型。

延迟优化方向

延迟高时，从以下角度排查：

排查项	方法	预期
Source 速度	`records-in` 指标	与上游生产速度匹配
Watermark 延迟	`currentOutputWatermark`	接近系统时间
反压	`backPressuredTimeMsPerSecond`	= 0
状态访问	RocksDB 指标	SST 文件数稳定、compaction 不持续排队
GC	JMX GC 指标	暂停时间与自己历史基线比较，不设固定毫秒阈值

小结

概念	API	用途
Counter	`.counter(name)`	计数
Gauge	`.gauge(name, supplier)`	当前值
Histogram	`.histogram(name, impl)`	分布统计
Meter	`.meter(name)`	速率
LatencyMarker	`setLatencyTrackingInterval()`	端到端延迟
MetricsReporter	`flink-conf.yaml`	导出到外部

下一节 8.2 反压与资源调优：反压检测、buffer 参数、slot 调优。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页