Apache Flinkでのスループットと待ち時間

Question

Apache Flink用の非常に簡単なJavaプログラムを作成しました。スループット（1秒あたり処理されるタプル数）や待ち時間（プログラムが毎回処理する必要がある時間など）入力タプル）。

https://ci.apache.org/projects/flink/flink-docs-release-1.2/monitoring/metrics.html

しかし、私は、私が欲しいものを得るためにそれらを使用する方法がわからないです：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 

env.readTextFile("/home/LizardKing/Documents/Power/Prova.csv") 
     .map(new MyMapper().writeAsCsv("/home/LizardKing/Results.csv"); 

JobExecutionResult res = env.execute(); 

私はFLINKは、いくつかの指標を公開することを知っています。リンクから、平均スループットを測定するために「メーター」を使用することができますが、定義した後、どのように使用すればよいですか？

Answer 1

私たちは、本物のストリーミングジョブで糸で動作するmeter、gaugeなどのカスタムメトリックを実行しています。私たちは、その後、MyMapperクラスにメーターを追加

バージョン1.2.1を使用している

<dependency> 
    <groupId>org.apache.flink</groupId> 
    <artifactId>flink-metrics-dropwizard</artifactId> 
    <version>${flink.version}</version> 
</dependency> 

追加の依存関係がのpom.xmlする：ここで

は手順です。

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

これが役に立ちます。