Apache Flink - Counter value displayed but meter values not displayed

Question

We are using Flink 1.8.0 and running it on EMR - Yarn and would like to measure the throughput.

1. 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();

    }
}

2. 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?

Answer 1

Turns out the meter displays whole number values and the rate is measured in decimals. When my load was a constant 1 event per second, it was actually measured as 0.9xxx something and hence was showing only 0 events per second.