12

I'm using Flink 1.4.1 and Beam 2.3.0, and would like to know is it possible to have metrics available in Flink WebUI (or anywhere at all), as in Dataflow WebUI ?

I've used counter like:

import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
...
Counter elementsRead = Metrics.counter(getClass(), "elements_read");
...
elementsRead.inc();

but I can't find "elements_read" counts available anywhere (Task Metrics or Accumulators) in Flink WebUI. I thought this will be straightforward after BEAM-773.

3 Answers 3

3
+50

Once you have selected a job in your dashboard, you will see the DAG for that job and below the DAG there are a list of tabs.

  1. Click on "Task Metrics" Tab
  2. Click on a box of your DAG
  3. Click on the Add Metric Button, to show that operator metric

Flink dashboard showing metrics

4
  • tried it, but no luck. My counter is not in the metrics list. How did you create your Beam counter/metrics ?
    – robosoul
    Mar 8, 2018 at 21:05
  • Humm... can you see your counter in the accumulators tab?
    – diegoreico
    Mar 9, 2018 at 7:29
  • @robosoul, any progress with that? I am also facing the same problem: all I can see is standard metrics and there is no sign of my custom ones.
    – Hermes
    Sep 18, 2019 at 9:58
  • @diegoreico .. I can see the metrics in Accumulators Tab but not in Metrics Tab.. I'm using Flink Version: 1.12.0.. With Latest Apache Beam Master Branch Code.. Jun 2, 2021 at 2:10
0

If your pipeline is running in detached mode, metrics are not supported to be queried. Refer this.

public class FlinkDetachedRunnerResult implements PipelineResult {

  FlinkDetachedRunnerResult() {}

  @Override
  public State getState() {
    return State.UNKNOWN;
  }

  @Override
  public MetricResults metrics() {
    throw new UnsupportedOperationException("The FlinkRunner does not currently support metrics.");
  }

  @Override
  public State cancel() throws IOException {
    throw new UnsupportedOperationException("Cancelling is not yet supported.");
  }

  @Override
  public State waitUntilFinish() {
    return State.UNKNOWN;
  }

  @Override
  public State waitUntilFinish(Duration duration) {
    return State.UNKNOWN;
  }

  @Override
  public String toString() {
    return "FlinkDetachedRunnerResult{}";
  }
}

However, I was able to view the metrics using slf4j reporter

1
  • @zorro how did you view the metrics with the slf4j reporter? May 31, 2019 at 15:38
0
from apache_beam.metrics.metric import Metrics
from apache_beam.metrics.metric import MetricsFilter
from apache_beam.options.pipeline_options import PipelineOptions
import apache_beam as beam
import csv
import logging

GAME_DATA = [
'user1_team1,team1,18,1447686663000,2015-11-16 15:11:03.921',
'user1_team1,team1,18,1447690263000,2015-11-16 16:11:03.921',
'user2_team2,team2,2,1447690263000,2015-11-16 16:11:03.955',
'user3_team3,team3,8,1447690263000,2015-11-16 16:11:03.955',
'user4_team3,team3,5,1447690263000,2015-11-16 16:11:03.959',
'user1_team1,team1,14,1447697463000,2015-11-16 18:11:03.955',
'robot1_team1,team1,9000,1447697463000,2015-11-16 18:11:03.955',
'robot2_team2,team2,1,1447697463000,2015-11-16 20:11:03.955',
'robot2_team2,team2,9000,1447697463000,2015-11-16 21:11:03.955',
'robot1_team1,1000,2447697463000,2915-11-16 21:11:03.955',
'robot2_team2,9000,1447697463000,2015-11-16 21:11:03.955']

class ParseGameEventFn(beam.DoFn):
    def __init__(self):
        super(ParseGameEventFn, self).__init__()
    self.game_events = Metrics.counter(self.__class__, 'game_events')

    def process(self, element, *args, **kwargs):
        try:
            self.game_events.inc()
            row = list(csv.reader([element]))[0]
            if int(row[2]) < 5:
               return
            yield {
                'user': row[0],
                'team': row[1],
                'score': int(row[2]),
                'timestamp': int(row[3]) / 1000.0,
            }
        except Exception as ex:
            logging.error('Parse error on {}: {}'.format(element, ex))

with beam.Pipeline(options=pipeline_options) as pipeline:
    results = (
        pipeline
        | "Create" >> beam.Create(GAME_DATA)
        | "Parsing" >> beam.ParDo(ParseGameEventFn())
        | "AddEventTimestamps" >> beam.Map(
             lambda elem: beam.window.TimestampedValue(elem, elem['timestamp']))
        | "Print" >> beam.Map(print))

metric_results = pipeline.result.metrics().query(MetricsFilter().with_name('game_events'))
outputs_user_counter = metric_results['counters'][0]
print(outputs_user_counter.committed)

Flink Configuration for Prometheus in conf/flink-conf.yaml

metrics.reporters: prom
metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
metrics.reporter.prom.port: 9250-9260

I can see the metrics in Accumulators Tab but not in Metrics Tab.. I'm using Flink Version: 1.12.0.. With Latest Apache Beam Master Branch Code..

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.