Apache Flink的吞吐量和延迟

我们正在运行在纱线上的生产流作业中运行自定义指标，例如仪表，仪表。

步骤如下：

对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;        }    }}

希望这可以帮助 。