我们正在运行在纱线上的生产流作业中运行自定义指标，例如仪表，仪表。
步骤如下：
对pom.xml的附加依赖
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-metrics-dropwizard</artifactId> <version>${flink.version}</version></dependency>我们正在使用1.2.1版
然后将仪表添加到MyMapper类。
import org.apache.flink.api.common.JobExecutionResult;import org.apache.flink.api.common.functions.RichMapFunction;import org.apache.flink.configuration.Configuration;import org.apache.flink.dropwizard.metrics.DropwizardMeterWrapper;import org.apache.flink.metrics.Meter;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class Test { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env .readTextFile("/home/LizardKing/documents/Power/Prova.csv") .map(new MyMapper()) .writeAsCsv("/home/LizardKing/Results.csv"); JobExecutionResult res = env.execute(); } private static class MyMapper extends RichMapFunction<String, Object> { private transient Meter meter; @Override public void open(Configuration parameters) throws Exception { super.open(parameters); this.meter = getRuntimeContext() .getMetricGroup() .meter("myMeter", new DropwizardMeterWrapper(new com.codahale.metrics.Meter())); } @Override public Object map(String value) throws Exception { this.meter.markEvent(); return value; } }}希望这可以帮助 。
