如何将flink用户自定义指标导出到prometheus和grafana

Question

我正在尝试根据this tutorial创建我的自定义指标变量>

通过提供示例代码，我可以获得事件和直方图。

我很困惑普罗米修斯和格拉法纳如何使用标识符。我还尝试稍微修改示例代码，但该指标不再起作用。

而且，我只能访问系统指标，而不能访问我自己的指标。

我的问题是：

1. 如何访问创建的计数器？例如counter1
2. metricGroup到底是什么？
3. 例如，我想检测一种模式从输入流中，并且在度量或仅将结果输出到时间序列数据库，例如influxdb？

预先感谢。

这里是地图功能

class FlinkMetricsExposingMapFunction extends RichMapFunction<SensorReading, SensorReading> {
  private static final long serialVersionUID = 1L;

  private transient Counter eventCounter;
  private transient Counter customCounter1;
  private transient Counter customCounter2;

  @Override
  public void open(Configuration parameters) {
    eventCounter = getRuntimeContext()
        .getMetricGroup().counter("events");

    customCounter1 = getRuntimeContext()
        .getMetricGroup()
        .addGroup("customCounterKey", "mod2")
        .counter("counter1");
    customCounter2 = getRuntimeContext()
        .getMetricGroup().addGroup("customCounterKey", "mod5").counter("counter2");



//    meter = getRuntimeContext().getMetricGroup().meter("eventMeter", new DropwizardMeterWrapper(dropwizardMeter));
  }

  @Override
  public SensorReading map(SensorReading value) {
    eventCounter.inc();
    if (value.getCurrTimestamp() % 2 == 0)
      customCounter1.inc();

    if (value.getCurrTimestamp() % 5 == 0)
      customCounter2.inc();

    if (value.getCurrTimestamp() % 2 == 0 && value.getCurrTimestamp() % 5 == 0)
      customCounter1.dec();
    return value;
  }
}

示例工作：

env
        .addSource(new SimpleSensorReadingGenerator())
        .name(SimpleSensorReadingGenerator.class.getSimpleName())
        .map(new FlinkMetricsExposingMapFunction())
        .name(FlinkMetricsExposingMapFunction.class.getSimpleName())
        .print()
        .name(DataStreamSink.class.getSimpleName());
更新

来自grafana的访问flink指标的屏幕截图：

flink-config.yaml

FROM flink:1.9.0
RUN echo "metrics.reporters: prom" >> "$FLINK_HOME/conf/flink-conf.yaml"; \
    echo "metrics.latency.interval: 1000" >> "$FLINK_HOME/conf/flink-conf.yaml"; \
    echo "metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter" >> "$FLINK_HOME/conf/flink-conf.yaml"; \
    mv $FLINK_HOME/opt/flink-metrics-prometheus-*.jar $FLINK_HOME/lib
COPY --from=builder /home/gradle/build/libs/*.jar $FLINK_HOME/lib/

教程中的默认地图功能：

 @Override
 public void open(Configuration parameters) {
    eventCounter = getRuntimeContext().getMetricGroup().counter("events");
    valueHistogram =
        getRuntimeContext()
            .getMetricGroup()
            .histogram("value_histogram", new DescriptiveStatisticsHistogram(10_000_000));
 }

我正在尝试根据本教程创建自定义指标变量，并提供了示例代码，可以获取事件和直方图。我很困惑...

Answer 1

1. 您创建的计数器可通过<system-scope>. customCounterKey.mod2.counter1访问。 <system-scope>在您的flink-conf.yaml中定义。如果未定义，则默认为<host>.taskmanager.<tm_id>.<job_name>.<operator_name>.<subtask_index>。