收集 flink flatmap 函数所用平均时间的指标

Question

我已经实现了一个 flink flatmap 函数，我想收集这个函数的平均时间指标，我计划通过 prometheus 监控它。

什么是好的方法呢？我在方法中添加了一个仪表（从 flink API 扩展仪表接口）。

public class SimpleGauge<T> implements Gauge<T> {

    private T mValue;

    @Override
    public T getValue() {
        return mValue;
    }

    public void setValue(T value){
        mValue = value;
    }
}

我从 flatmap 函数中调用 setValue：

float endTime = (System.currentTimeMillis() - startTime) / 1000F;
this.gauge.setValue(endTime);

它会起作用吗？

Answer 1

它是否有效取决于您如何创建Gauge。如果您使用的是普通的FlatMapFunction，您可能应该切换到RichFlatMapFunction，然后初始化并注册您的仪表getRuntimeContext。

另外，您可能想了解一下指标类型，因为在这种情况下，Histogram 似乎比 Gauge 更好，因为在大多数情况下，测量平均延迟通常不是最好的主意。