为何Flink Dashboard未展示源端接收与Sink端写入的记录数？

我之前也遇到过一模一样的问题，一开始误以为作业没在正常运行，后来才搞清楚这和Flink的指标收集机制直接相关。下面给你拆解具体原因和对应的解决方式：

为什么会显示0？ #

• 自定义算子未集成指标统计：如果你的Source或Sink是自己实现的，默认情况下Flink不会自动帮你统计记录数——它依赖算子主动通过MetricGroup上报指标。要是代码里没加这个逻辑，Dashboard自然拿不到数据，只能显示0。
• 内置算子被禁用指标：有些内置的Source/Sink（比如Kafka、JDBC）支持通过配置关闭指标收集，如果不小心开启了disableMetrics()这类选项，或者全局配置里禁用了相关指标，也会出现数值为0的情况。
• 序列化逻辑外置导致计数缺失：如果你的Sink把序列化Schema放到了前面的算子（比如Map）里处理，Sink本身没有接触到原始记录的计数逻辑，也会让Dashboard里的Sink记录数显示0。
• （补充）极少数情况下，作业刚启动时指标还没完成第一次上报，或者窗口/背压导致统计延迟，但这种一般是暂时的，你说的持续显示0大概率不是这个原因。

怎么让它显示真实数值？ #

针对自定义Source/Sink #

手动在算子代码里集成指标统计是最直接的方法，核心是通过RuntimeContext获取MetricGroup并创建计数器：

// 以自定义Sink为例，在open方法初始化计数器
private Counter sinkRecordCounter;

@Override
public void open(Configuration parameters) throws Exception {
    super.open(parameters);
    // 创建名为sink-record-count的计数器，会自动上报到Dashboard
    sinkRecordCounter = getRuntimeContext().getMetricGroup().counter("sink-record-count");
}

@Override
public void invoke(T record, Context context) throws Exception {
    // 你的业务处理逻辑
    processRecord(record);
    // 每处理一条记录，计数器+1
    sinkRecordCounter.inc();
}

自定义Source的逻辑类似，在run()方法里每读取一条记录就调用计数器的inc()方法即可。

针对内置Source/Sink #

• 检查配置：比如Kafka Sink要确认没有设置metrics.disable=true，可以在配置里明确开启记录数相关指标，或者直接开启所有指标。
• 版本排查：某些旧版本的Flink在特定场景下（比如使用了特定序列化Schema）存在指标统计的bug，升级到稳定的新版本（比如1.17+）可能会解决问题。

针对Flink SQL场景 #

如果是用SQL编写的作业，需要确保开启了指标收集，在SQL配置里添加：

SET table.exec.metrics.enabled = true;

这样SQL自动生成的Source/Sink算子会主动上报记录数指标。

序列化Schema相关的特殊处理 #

如果你的序列化逻辑必须放在Sink之外，那可以考虑在序列化后的算子里添加计数器，但这样Dashboard里的Sink指标还是不会显示。更推荐的方式是把计数逻辑移到Sink内部，哪怕只是在序列化完成后做一次计数，这样才能让Dashboard的Sink指标显示真实数值。

内容的提问来源于stack exchange，提问作者Jicaar