We are using Flink 1.8.0 and running it on EMR - Yarn and would like to measure the throughput.
- Because our operators are chained, we have added meters and counters in our code - essentially an async operator that makes API calls with kinesis as both source and sync. In the Application Master i.e. Flink's web UI, we are able to get the value for the counters but not the meters.
public class AsyncClass extends RichAsyncFunction<String, String> {
private transient Counter counter;
private transient Meter meter;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
this.counter = getRuntimeContext()
.getMetricGroup()
.counter("myCounter");
this.meter = getRuntimeContext()
.getMetricGroup()
.meter("myMeter", new DropwizardMeterWrapper(new com.codahale.metrics.Meter()));
}
@Override
public void close() throws Exception {
super.close();
ExecutorUtils.gracefulShutdown(20000, TimeUnit.MILLISECONDS, executorService);
}
@Override
public void asyncInvoke(String key, final ResultFuture<String> resultFuture) throws Exception {
resultFuture.complete(key);
this.meter.markEvent();
this.counter.inc();
}
}
- To measure the complete throughput of the application, we obviously need the throughput of all the task managers together. Using meters, we are able to get the metrics for individual task managers. Is there any way to measure it at the operator level?
