flink内部实现了一套metric数据收集库。 同时flink自身系统有一些固定的metric数据， 包括系统的一些指标，CPU,内存, IO 或者各个task运行的一些指标。具体包含那些指标可以查看官方文档: flink-metric
同时我们也可以利用系统的metric库在自己的代码中进行打点收集metrics数据。此外, flink提供了外部接口，可以用来导出这些metrics数据.
在官方的文档中有介绍, 需要继承Richfunction 才能获得对应的metric对象， 用法如下:
public class MyMapper extends RichMapFunction<String, String> {
private transient Counter counter;
@Override
public void open(Configuration config) {
this.counter = getRuntimeContext()
.getMetricGroup()
.counter("myCounter");
}
@Override
public String map(String value) throws Exception {
this.counter.inc();
return value;
}
}
在flink中, 提供了方便的metric数据导出的库，通过实现自己的reporter，可以将metrics数据导出到不同的系统.
官方提供有多种reporter库，JMX, Graphite, Slf4j... 等等. 同时，我们可以自定义实现metric库，来导入到自己的系统.
自定义reporter类
实现MetricReporter类中的open,close, notifyOfAddedMetric, notifyOfRemovedMetric方法
实现Scheduled的report方法 ，在刚方法中实现写入到其他系统的逻辑
实现CharacterFilter的 filterCharacters方法, 用于对scope进行过滤.
public class FalconReporter implements MetricReporter, CharacterFilter, Scheduled {
private static final Logger LOG =LoggerFactory.getLogger(FalconReporter.class);
private final Map<Gauge<?>, MetricTag> gauges = new ConcurrentHashMap<>();
private final Map<Counter, MetricTag> counters = new ConcurrentHashMap<>();
private final Map<Histogram, MetricTag> histograms = new ConcurrentHashMap<>();
private final Map<Meter, MetricTag> meters = new ConcurrentHashMap<>();
@Override
public String filterCharacters(String s) {
return s;
}
@Override
public void open(MetricConfig metricConfig) {
}
@Override
public void close() {
}
@Override
public void notifyOfAddedMetric(Metric metric, String s, MetricGroup metricGroup) {
}
@Override
public void notifyOfRemovedMetric(Metric metric, String s, MetricGroup metricGroup) {
}
@Override
public void report() {
}
}
配置reporter
在flink-conf.yaml中配置即可，配置如下
metrics.reporters: slf4j, jmx
metrics.reporter.slf4j.class: org.apache.flink.metrics.falcon.FalconReporter
metrics.reporter.slf4j.interval: 60 SECONDS
metrics.reporters 用于配置类型名， 自定义即可
metrics.reporter.slf4j.class: 配置对应类型的reporter类
metrics.reporter.slf4j.interval: 60 SECONDS 消息上报的间隔
metrics.reporter.slf4j.* 可以自定义配置， 可以在open(MetricConfig metricConfig) 中的获得对应的config