Flink statsD метрика - PullRequest
Новая
       10

Flink statsD метрика

0 голосов
/ 28 мая 2018

Я не могу увидеть статистику, которая была отправлена ​​с flink на statsd.Уже размещены flink-metrics-statsd-1.4.1.jar в папках JM и TMs / lib.С кодом ниже, я бы ожидал увидеть «myCounter» как серию effxDB.Но в притоке нет ничего, так что ничего не показывай в графане.Любой совет?спасибо! 

public class StatsdCounter extends RichMapFunction<String, String> {
private transient Counter counter;

@Override
public void open(Configuration config) {
    this.counter = getRuntimeContext()
    .getMetricGroup()
    .counter("myCounter");
}

@Override
public String map(String value) throws Exception {
    this.counter.inc();
    return value;
}

в flink-conf.yaml, 

metrics.scope.jm: <host>.jobmanager
metrics.scope.jm.job: <host>.jobmanager.<job_name>
metrics.scope.tm: <host>.taskmanager.<tm_id>
metrics.scope.tm.job: <host>.taskmanager.<tm_id>.<job_name>
metrics.scope.task: <host>.taskmanager.<tm_id>.<job_name>.<task_name>.<subtask_index>
metrics.scope.operator: <host>.taskmanager.<tm_id>.<job_name>.<operator_name>.<subtask_index>

metrics.reporters: stsd
metrics.reporter.stsd.class: org.apache.flink.metrics.statsd.StatsDReporter
metrics.reporter.stsd.host: 192.168.56.10
metrics.reporter.stsd.port: 8127
metrics.reporter.stsd.interval: 1 SECONDS

Но это будет работать, если мы установим statsd вручную в коде, 

public void open(Configuration configuration) throws Exception {
    super.open(configuration);
    counter = new AtomicInteger(); 
    stats = new NonBlockingStatsDClient(xxx);
}

@Override
public String map(String value) throws Exception { 
  stats.increment(statsdAspect); 
  return value;
}
...