Flink Metric Reporters 实战统一配置模型、过滤规则、Push/Pull、Tags/Identifier 与常用 Reporter 模板

1. Reporter 是怎么工作的？

• 在flink-conf.yaml配置metrics.reporters指定启用哪些 reporter
• 每个 reporter 都要配置metrics.reporter.<name>.factory.class
• Push 型 reporter 还可以配置metrics.reporter.<name>.interval以控制上报周期
• Reporter jar 必须在 Flink 启动时可见；Reporter以插件方式加载（Flink 文档中的 reporter 通常默认可用）

2. 两个关键维度：Identifier vs Tags，Push vs Pull

2.1 Identifier-based（标识符型）

把 scope 信息和指标名拼成一个扁平字符串，例如：

• job.MyJobName.numRestarts

典型：Graphite、StatsD、Slf4j（偏 identifier）

2.2 Tag-based（标签型）

把“逻辑指标类”与“实例标签”分开，例如：

• 逻辑指标：job.numRestarts
• 标签：jobName=MyJobName

典型：Prometheus、InfluxDB、Datadog（偏 tags）

2.3 Push vs Pull

• Push：Reporter 定期主动发给外部系统（需要 interval）
• Pull：外部系统来抓取/查询（Prometheus/JMX 常见）

3. 通用配置参数（所有 reporter 都能用的那套）

配置格式统一为：

• metrics.reporter.<reporter_name>.<property>

通用 Key（最常用的几项）：

• factory.class：必配，指定 ReporterFactory
• interval：默认 10s，仅 Push 型生效
• scope.delimiter：拼接 metric identifier 的分隔符（默认.）
• scope.variables.additional：给 tag 型 reporter 增加额外标签（map）
• scope.variables.excludes：排除某些变量（tag 型）
• filter.includes/filter.excludes：指标过滤（强烈建议用来控量）

3.1 多 reporter 并存示例

metrics.reporters:my_jmx_reporter,my_other_reportermetrics.reporter.my_jmx_reporter.factory.class:org.apache.flink.metrics.jmx.JMXReporterFactorymetrics.reporter.my_jmx_reporter.port:9020-9040metrics.reporter.my_other_reporter.factory.class:org.apache.flink.metrics.graphite.GraphiteReporterFactorymetrics.reporter.my_other_reporter.host:192.168.1.1metrics.reporter.my_other_reporter.port:10000

4. 过滤器（filter.includes/excludes）：写对了能省一大半成本

过滤器格式（一个字符串就是一个 filter）：

<scope>[:<name>[,<name>][:<type>[,<type>]]]

• scope：逻辑范围（用.分段，*通配）
• name：指标名模式（逗号分隔，*通配）
• type：指标类型（counter,meter,gauge,histogram）

匹配规则：
一个指标命中 filter 需要同时满足：

• scope 匹配
• name 至少一个模式匹配
• type 至少一个类型匹配

4.1 常见例子（你直接拿去改）

• 全局匹配某类指标名：

metrics.reporter.prom.filter.includes:*:numRecords*

• 只要 operator 层的 numRecords 指标：

metrics.reporter.prom.filter.includes:*.job.task.operator:numRecords*

• 只要 operator 层的 meter（如 numRecordsInPerSecond）：

metrics.reporter.prom.filter.includes:*.job.task.operator:numRecords*:meter

• 只要 Records/Bytes 相关的 counter + meter：

metrics.reporter.prom.filter.includes:*:*Records*,*Bytes*:counter,meter

生产建议：
Prometheus/Datadog/InfluxDB 这类 tags 报表很容易“标签爆炸”，务必用过滤器控制输出范围，否则指标量和成本会快速失控。

5. 常用 Reporter 配置模板与注意点

下面把文档里几个最常用的 reporter 配置按“可直接复制”方式整理出来。

5.1 JMX（pull / tags）

适合 JVM 生态排障、用 JConsole/JMC 或被采集系统抓取。

metrics.reporter.jmx.factory.class:org.apache.flink.metrics.jmx.JMXReporterFactorymetrics.reporter.jmx.port:9250-9260

建议用端口范围：同一台机器上可能同时有 JM/TM，避免端口冲突；实际绑定端口会在日志里打印。

5.2 Prometheus（pull / tags）

最常用的云原生采集方式。

metrics.reporter.prom.factory.class:org.apache.flink.metrics.prometheus.PrometheusReporterFactorymetrics.reporter.prom.port:9250-9260# 可选：标签值字符过滤（默认 true）# metrics.reporter.prom.filterLabelValueCharacters: true

类型映射注意点（很实用）：

• Flink Counter → Prometheus Gauge（因为 Prometheus counter 不允许递减）
• Histogram → Summary（带固定 quantiles）
• Meter → Gauge（输出 rate）

5.3 Prometheus PushGateway（push / tags）

适合“短生命周期任务”或无法被 Prometheus 直连抓取的环境。

metrics.reporter.promgateway.factory.class:org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterFactorymetrics.reporter.promgateway.hostUrl:http://localhost:9091metrics.reporter.promgateway.jobName:myJobmetrics.reporter.promgateway.randomJobNameSuffix:truemetrics.reporter.promgateway.deleteOnShutdown:falsemetrics.reporter.promgateway.groupingKey:k1=v1;k2=v2metrics.reporter.promgateway.interval:60 SECONDS

5.4 InfluxDB（push / tags）

适合时序库落库分析。

metrics.reporter.influxdb.factory.class:org.apache.flink.metrics.influxdb.InfluxdbReporterFactorymetrics.reporter.influxdb.scheme:httpmetrics.reporter.influxdb.host:localhostmetrics.reporter.influxdb.port:8086metrics.reporter.influxdb.db:flinkmetrics.reporter.influxdb.username:flink-metricsmetrics.reporter.influxdb.password:qwertymetrics.reporter.influxdb.retentionPolicy:one_hourmetrics.reporter.influxdb.consistency:ANYmetrics.reporter.influxdb.connectTimeout:60000metrics.reporter.influxdb.writeTimeout:60000metrics.reporter.influxdb.interval:60 SECONDS

5.5 Graphite（push / identifier）

metrics.reporter.grph.factory.class:org.apache.flink.metrics.graphite.GraphiteReporterFactorymetrics.reporter.grph.host:localhostmetrics.reporter.grph.port:2003metrics.reporter.grph.protocol:TCPmetrics.reporter.grph.interval:60 SECONDS

5.6 StatsD（push / identifier）

metrics.reporter.stsd.factory.class:org.apache.flink.metrics.statsd.StatsDReporterFactorymetrics.reporter.stsd.host:localhostmetrics.reporter.stsd.port:8125metrics.reporter.stsd.interval:60 SECONDS

5.7 Datadog（push / tags，兼有 identifier）

Datadog 会把 host/job_name/tm_id/subtask_index/task_name/operator_name 等作为 tags 上报。

metrics.reporter.dghttp.factory.class:org.apache.flink.metrics.datadog.DatadogHttpReporterFactorymetrics.reporter.dghttp.apikey:xxxmetrics.reporter.dghttp.proxyHost:my.web.proxy.commetrics.reporter.dghttp.proxyPort:8080metrics.reporter.dghttp.dataCenter:USmetrics.reporter.dghttp.maxMetricsPerRequest:2000metrics.reporter.dghttp.interval:60 SECONDSmetrics.reporter.dghttp.useLogicalIdentifier:true

注意：Histogram 会以一组 gauge 的形式上报（如<metric>.<aggregation>），并且聚合并非基于每个上报周期重新计算的那种“区间聚合”，理解这一点很重要。

5.8 OpenTelemetry（Otel）

适合统一接入观测体系（Collector → 多后端）。

metrics.reporter.otel.factory.class:org.apache.flink.metrics.otel.OpenTelemetryMetricReporterFactorymetrics.reporter.otel.exporter.endpoint:http://127.0.0.1:1337metrics.reporter.otel.exporter.protocol:gRPC# 可选：# metrics.reporter.otel.exporter.timeout: 10s# metrics.reporter.otel.service.name: flink# metrics.reporter.otel.service.version: 1.0.0

HTTP 协议的写法：

metrics.reporter.otel.factory.class:org.apache.flink.metrics.otel.OpenTelemetryMetricReporterFactorymetrics.reporter.otel.exporter.endpoint:http://127.0.0.1:9090metrics.reporter.otel.exporter.protocol:HTTP

5.9 Slf4j（push / identifier）

适合调试或小规模环境，不建议大集群长期打开（日志量会爆）。

metrics.reporter.slf4j.factory.class:org.apache.flink.metrics.slf4j.Slf4jReporterFactorymetrics.reporter.slf4j.interval:60 SECONDS

6. 自定义 Reporter：两条接口要点

• 实现org.apache.flink.metrics.reporter.MetricReporter
• 如果要定期上报，实现Scheduled接口
• report()方法不要长时间阻塞，耗时操作建议异步化
• 再实现MetricReporterFactory，就能以插件方式加载（也更符合 Flink 插件隔离机制）

7. 一套“生产默认建议”（不引战但很实用）

• 云原生/K8s：优先 Prometheus（pull/tags），端口用范围，配 filter 控量
• 需要落库分析：InfluxDB/Otel → 统一收敛
• 遗留体系：Graphite/StatsD 仍可用，但 identifier 模式要注意命名层级规划
• 任何 tags 系统：谨防标签爆炸（scope.variables + filter 是关键）