Apache Beam计数器/指标在Flink WebUI中不可用。

Question

我正在使用Flink 1.4.1和Beam 2.3.0，想知道是否可以像Dataflow WebUI那样，在Flink WebUI（或其他任何地方）中使用度量指标？

我已经使用了计数器：

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

但我在Flink WebUI中找不到"elements_read"计数器的数据（无论是任务度量还是累加器）。在BEAM-773修复之后，我认为这应该很简单。

- robosoul

Answer 1

一旦您在仪表板中选择了一个作业，您将看到该作业的DAG图，下面是一系列选项卡。

1. 点击“任务指标”选项卡
2. 点击DAG图中的一个框
3. 单击“添加指标”按钮，以显示该运算符的指标

- diegoreico

---

我尝试了一下，但没有成功。我的计数器不在指标列表中。你是如何创建你的Beam计数器/指标的？ - robosoul

嗯...你能在累加器选项卡中看到计数器吗？ - diegoreico

@robosoul，有进展吗？我也遇到了同样的问题：我只能看到标准指标，没有我的自定义指标的迹象。 - Hermes

@diegoreico 我可以在累加器选项卡中看到指标，但无法在指标选项卡中看到。 我正在使用Flink版本：1.12.0。使用最新的Apache Beam主分支代码。 - Ravi D. Borse

Answer 2

如果您的管道在分离模式下运行，则不支持查询指标。请参见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{}";
  }
}

然而，我能够使用 slf4j reporter 查看指标。

- zorro

---

@zorro，你如何使用slf4j报告器查看指标？ - autodidacticon

Answer 3

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)

在 conf/flink-conf.yaml 中配置 Flink 与 Prometheus

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

我可以在累加器选项卡中看到指标，但无法在指标选项卡中看到。我正在使用Flink版本：1.12.0。使用最新的Apache Beam主分支代码。

- Ravi D. Borse