This documentation is for an unreleased version of Apache Flink. We recommend you use the latest stable version.

メトリクス #

Flinkは、メトリクスを収集し、外部システムに公開するメトリックシステムを公開します。

メトリクスの登録 #

You can access the metric system from any user function that extends RichFunction by calling getRuntimeContext().getMetricGroup(). This method returns a MetricGroup object on which you can create and register new metrics.

メトリクスの型 #

Flink supports Counters, Gauges, Histograms and Meters.

カウンター #

カウンターは何かを数えるために使われます。The current value can be in- or decremented using inc()/inc(long n) or dec()/dec(long n). You can create and register a Counter by calling counter(String name) on a MetricGroup.

public class MyMapper 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;
  }
}

class MyMapper extends RichMapFunction[String,String] {
  @transient private var counter: Counter = _

  override def open(parameters: Configuration): Unit = {
    counter = getRuntimeContext()
      .getMetricGroup()
      .counter("myCounter")
  }

  override def map(value: String): String = {
    counter.inc()
    値
  }
}

class MyMapper(MapFunction):
    def __init__(self):
        self.counter = None

    def open(self, runtime_context: RuntimeContext):
        self.counter = runtime_context \
            .get_metrics_group() \
            .counter("my_counter")

    def map(self, value: str):
        self.counter.inc()
        return value

Alternatively you can also use your own Counter implementation:

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

  @Override
  public void open(Configuration config) {
    this.counter = getRuntimeContext()
      .getMetricGroup()
      .counter("myCustomCounter", new CustomCounter());
  }

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

class MyMapper extends RichMapFunction[String,String] {
  @transient private var counter: Counter = _

  override def open(parameters: Configuration): Unit = {
    counter = getRuntimeContext()
      .getMetricGroup()
      .counter("myCustomCounter", new CustomCounter())
  }

  override def map(value: String): String = {
    counter.inc()
    値
  }
}

また、Python APIではサポートされません。

ゲージ #

A Gauge provides a value of any type on demand. In order to use a Gauge you must first create a class that implements the org.apache.flink.metrics.Gauge interface. 帰り値の型についての制限はありません。 You can register a gauge by calling gauge(String name, Gauge gauge) on a MetricGroup.

public class MyMapper extends RichMapFunction<String, String> {
  private transient int valueToExpose = 0;

  @Override
  public void open(Configuration config) {
    getRuntimeContext()
      .getMetricGroup()
      .gauge("MyGauge", new Gauge<Integer>() {
        @Override
        public Integer getValue() {
          return valueToExpose;
        }
      });
  }

  @Override
  public String map(String value) throws Exception {
    valueToExpose++;
    return value;
  }
}

new class MyMapper extends RichMapFunction[String,String] {
  @transient private var valueToExpose = 0

  override def open(parameters: Configuration): Unit = {
    getRuntimeContext()
      .getMetricGroup()
      .gauge[Int, ScalaGauge[Int]]("MyGauge", ScalaGauge[Int]( () => valueToExpose ) )
  }

  override def map(value: String): String = {
    valueToExpose += 1
    値
  }
}

class MyMapper(MapFunction):
    def __init__(self):
        self.value_to_expose = 0

    def open(self, runtime_context: RuntimeContext):
        runtime_context \
            .get_metrics_group() \
            .gauge("my_gauge", lambda: self.value_to_expose)

    def map(self, value: str):
        self.value_to_expose += 1
        return value

Note that reporters will turn the exposed object into a String, which means that a meaningful toString() implementation is required.

ヒストグラム #

A Histogram measures the distribution of long values. You can register one by calling histogram(String name, Histogram histogram) on a MetricGroup.

public class MyMapper extends RichMapFunction<Long, Long> {
  private transient Histogram histogram;

  @Override
  public void open(Configuration config) {
    this.histogram = getRuntimeContext()
      .getMetricGroup()
      .histogram("myHistogram", new MyHistogram());
  }

  @Override
  public Long map(Long value) throws Exception {
    this.histogram.update(value);
    return value;
  }
}

class MyMapper extends RichMapFunction[Long,Long] {
  @transient private var histogram: Histogram = _

  override def open(parameters: Configuration): Unit = {
    histogram = getRuntimeContext()
      .getMetricGroup()
      .histogram("myHistogram", new MyHistogram())
  }

  override def map(value: Long): Long = {
    histogram.update(value)
    値
  }
}

また、Python APIではサポートされません。

Flink does not provide a default implementation for Histogram, but offers a Wrapper that allows usage of Codahale/DropWizard histograms. To use this wrapper add the following dependency in your pom.xml:

<dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-metrics-dropwizard</artifactId>
      <version>1.19-SNAPSHOT</version>
</dependency>

そして、以下のように Codahale/DropWizard ヒストグラムを登録することができます。

public class MyMapper extends RichMapFunction<Long, Long> {
  private transient Histogram histogram;

  @Override
  public void open(Configuration config) {
    com.codahale.metrics.Histogram dropwizardHistogram =
      new com.codahale.metrics.Histogram(new SlidingWindowReservoir(500));

    this.histogram = getRuntimeContext()
      .getMetricGroup()
      .histogram("myHistogram", new DropwizardHistogramWrapper(dropwizardHistogram));
  }
  
  @Override
  public Long map(Long value) throws Exception {
    this.histogram.update(value);
    return value;
  }
}

class MyMapper extends RichMapFunction[Long, Long] {
  @transient private var histogram: Histogram = _

  override def open(config: Configuration): Unit = {
    val dropwizardHistogram =
      new com.codahale.metrics.Histogram(new SlidingWindowReservoir(500))
        
    histogram = getRuntimeContext()
      .getMetricGroup()
      .histogram("myHistogram", new DropwizardHistogramWrapper(dropwizardHistogram))
  }
  
  override def map(value: Long): Long = {
    histogram.update(value)
    値
  }
}

また、Python APIではサポートされません。

メーター #

A Meter measures an average throughput. An occurrence of an event can be registered with the markEvent() method. Occurrence of multiple events at the same time can be registered with markEvent(long n) method. You can register a meter by calling meter(String name, Meter meter) on a MetricGroup.

public class MyMapper extends RichMapFunction<Long, Long> {
  private transient Meter meter;

  @Override
  public void open(Configuration config) {
    this.meter = getRuntimeContext()
      .getMetricGroup()
      .meter("myMeter", new MyMeter());
  }

  @Override
  public Long map(Long value) throws Exception {
    this.meter.markEvent();
    return value;
  }
}

class MyMapper extends RichMapFunction[Long,Long] {
  @transient private var meter: Meter = _

  override def open(config: Configuration): Unit = {
    meter = getRuntimeContext()
      .getMetricGroup()
      .meter("myMeter", new MyMeter())
  }

  override def map(value: Long): Long = {
    meter.markEvent()
    値
  }
}

class MyMapperMeter(MapFunction):
    def __init__(self):
        self.meter = None

    def open(self, runtime_context: RuntimeContext):
        # an average rate of events per second over 120s, default is 60s.
        self.meter = runtime_context
            .get_metrics_group()
            .meter("my_meter", time_span_in_seconds=120)

    def map(self, value: str):
        self.meter.mark_event()
        return value

Flink offers a Wrapper that allows usage of Codahale/DropWizard meters. To use this wrapper add the following dependency in your pom.xml:

<dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-metrics-dropwizard</artifactId>
      <version>1.19-SNAPSHOT</version>
</dependency>

以下のように Codahale/DropWizard meter を登録することができます:

public class MyMapper extends RichMapFunction<Long, Long> {
  private transient Meter meter;

  @Override
  public void open(Configuration config) {
    com.codahale.metrics.Meter dropwizardMeter = new com.codahale.metrics.Meter();

    this.meter = getRuntimeContext()
      .getMetricGroup()
      .meter("myMeter", new DropwizardMeterWrapper(dropwizardMeter));
  }

  @Override
  public Long map(Long value) throws Exception {
    this.meter.markEvent();
    return value;
  }
}

class MyMapper extends RichMapFunction[Long,Long] {
  @transient private var meter: Meter = _

  override def open(config: Configuration): Unit = {
    val dropwizardMeter: com.codahale.metrics.Meter = new com.codahale.metrics.Meter()
  
    meter = getRuntimeContext()
      .getMetricGroup()
      .meter("myMeter", new DropwizardMeterWrapper(dropwizardMeter))
  }

  override def map(value: Long): Long = {
    meter.markEvent()
    値
  }
}

また、Python APIではサポートされません。

スコープ #

各メトリックは識別子とメトリックが報告されるキー-値のペアのセットが割り当てられます。

識別子は3つのコンポーネントに基づきます: メトリックを登録する時のユーザ定義名、任意のユーザ定義のスコープ、およびシステムが提供するスコープ。 For example, if A.B is the system scope, C.D the user scope and E the name, then the identifier for the metric will be A.B.C.D.E.

You can configure which delimiter to use for the identifier (default: .) by setting the metrics.scope.delimiter key in conf/flink-conf.yaml.

ユーザ スコープ #

You can define a user scope by calling MetricGroup#addGroup(String name), MetricGroup#addGroup(int name) or MetricGroup#addGroup(String key, String value). These methods affect what MetricGroup#getMetricIdentifier and MetricGroup#getScopeComponents return.

counter = getRuntimeContext()
  .getMetricGroup()
  .addGroup("MyMetrics")
  .counter("myCounter");

counter = getRuntimeContext()
  .getMetricGroup()
  .addGroup("MyMetricsKey", "MyMetricsValue")
  .counter("myCounter");

counter = getRuntimeContext()
  .getMetricGroup()
  .addGroup("MyMetrics")
  .counter("myCounter")

counter = getRuntimeContext()
  .getMetricGroup()
  .addGroup("MyMetricsKey", "MyMetricsValue")
  .counter("myCounter")

counter = runtime_context \
    .get_metric_group() \
    .add_group("my_metrics") \
    .counter("my_counter")

counter = runtime_context \
    .get_metric_group() \
    .add_group("my_metrics_key", "my_metrics_value") \
    .counter("my_counter")

システム スコープ #

システムスコープはメトリックについてのコンテキスト情報を含みます。例えば、どのタスク内で登録されたか、あるいはタスクが何のジョブに所属しているか。

Which context information should be included can be configured by setting the following keys in conf/flink-conf.yaml. Each of these keys expect a format string that may contain constants (e.g. “taskmanager”) and variables (e.g. “<task_id>”) which will be replaced at runtime.

• metrics.scope.jm
  • Default: <host>.jobmanager
  • ジョブマネージャーにスコープされていた全てのメトリックに適用されます。
• metrics.scope.jm-job
  • Default: <host>.jobmanager.<job_name>
  • ジョブマネージャーとジョブにスコープされていた全てのメトリックに適用されます。
• metrics.scope.tm
  • Default: <host>.taskmanager.<tm_id>
  • タスクマネージャーにスコープされていた全てのメトリックに適用されます。
• metrics.scope.tm-job
  • Default: <host>.taskmanager.<tm_id>.<job_name>
  • タスクマネージャーおよびジョブにスコープされていた全てのメトリックに適用されます。
• metrics.scope.task
  • Default: <host>.taskmanager.<tm_id>.<job_name>.<task_name>.<subtask_index>
  • タスクにスコープされていた全てのメトリックに適用されます。
• metrics.scope.operator
  • Default: <host>.taskmanager.<tm_id>.<job_name>.<operator_name>.<subtask_index>
  • オペレータにスコープされていた全てのメトリックに適用されます。

変数の数あるいは順番に制限はありません。変数は大文字小文字を区別します。

The default scope for operator metrics will result in an identifier akin to localhost.taskmanager.1234.MyJob.MyOperator.0.MyMetric

タスク名を含めたいがタスクマネージャーの情報を省略したい場合は、以下の形式を指定することもできます:

metrics.scope.operator: <host>.<job_name>.<task_name>.<operator_name>.<subtask_index>

This could create the identifier localhost.MyJob.MySource_->_MyOperator.MyOperator.0.MyMetric.

Note that for this format string an identifier clash can occur should the same job be run multiple times concurrently, which can lead to inconsistent metric data. As such it is advised to either use format strings that provide a certain degree of uniqueness by including IDs (e.g <job_id>) or by assigning unique names to jobs and operators.

全ての変数のリスト #

• JobManager: <host>
• TaskManager: <host>, <tm_id>
• Job: <job_id>, <job_name>
• Task: <task_id>, <task_name>, <task_attempt_id>, <task_attempt_num>, <subtask_index>
• Operator: <operator_id>,<operator_name>, <subtask_index>

Important: For the Batch API, <operator_id> is always equal to <task_id>.

ユーザ変数 #

You can define a user variable by calling MetricGroup#addGroup(String key, String value). This method affects what MetricGroup#getMetricIdentifier, MetricGroup#getScopeComponents and MetricGroup#getAllVariables() returns.

重要: ユーザ変数はスコープ形式では使えません。

counter = getRuntimeContext()
  .getMetricGroup()
  .addGroup("MyMetricsKey", "MyMetricsValue")
  .counter("myCounter");

counter = getRuntimeContext()
  .getMetricGroup()
  .addGroup("MyMetricsKey", "MyMetricsValue")
  .counter("myCounter")

Reporter #

For information on how to set up Flink’s metric reporters please take a look at the metric reporters documentation.

システム メトリクス #

デフォルトでは、Flinkは現在の状態の深い洞察を提供する幾つかのメトリクスを集めます。 この章はこれら全てのメトリクスのリファレンスです。

以下の表は一般的に5つのカラムを特色とします:

• The “Scope” column describes which scope format is used to generate the system scope. For example, if the cell contains “Operator” then the scope format for “metrics.scope.operator” is used. If the cell contains multiple values, separated by a slash, then the metrics are reported multiple times for different entities, like for both job- and taskmanagers.

• The (optional)“Infix” column describes which infix is appended to the system scope.

• The “Metrics” column lists the names of all metrics that are registered for the given scope and infix.

• The “Description” column provides information as to what a given metric is measuring.

• The “Type” column describes which metric type is used for the measurement.

Note that all dots in the infix/metric name columns are still subject to the “metrics.delimiter” setting.

従って、メトリックの識別子を推測するには:

1. Take the scope-format based on the “Scope” column
2. Append the value in the “Infix” column if present, and account for the “metrics.delimiter” setting
3. メトリック名を追加する。

CPU #

メモリ #

The memory-related metrics require Oracle’s memory management (also included in OpenJDK’s Hotspot implementation) to be in place. Some metrics might not be exposed when using other JVM implementations (e.g. IBM’s J9).

スレッド #

GarbageCollection #

ClassLoader #

ネットワーク #

Deprecated: use Default shuffle service metrics

デフォルトのシャッフル サービス #

nettyネットワーク通信を使用したタスク エグゼキュータ間のデータ交換に関係するメトリクス。

クラスタ #

可用性 #

The metrics in this table are available for each of the following job states: INITIALIZING, CREATED, RUNNING, RESTARTING, CANCELLING, FAILING. Whether these metrics are reported depends on the metrics.job.status.enable setting.

Evolving The semantics of these metrics may change in later releases.

Experimental

While the job is in the RUNNING state the metrics in this table provide additional details on what the job is currently doing. Whether these metrics are reported depends on the metrics.job.status.enable setting.

スコープ	メトリクス	説明	種類
Job (only available on JobManager)	deployingState	Return 1 if the job is currently deploying* tasks, otherwise return 0.	ゲージ
	deployingTime	Return the time (in milliseconds) since the job has started deploying* tasks, otherwise return 0.	ゲージ
	deployingTimeTotal	Return how much time (in milliseconds) the job has spent deploying* tasks in total.	ゲージ

*A job is considered to be deploying tasks when:

• for streaming jobs, any task is in the DEPLOYING state
• for batch jobs, if at least 1 task is in the DEPLOYING state, and there are no INITIALIZING/RUNNING tasks

チェックポイント #

Note that for failed checkpoints, metrics are updated on a best efforts basis and may be not accurate.

State Access Latency #

RocksDB #

Certain RocksDB native metrics are available but disabled by default, you can find full documentation here

State Changelog #

Note that the metrics are only available via reporters.

IO #

コネクタ #

Kafka コネクタ #

Please refer to Kafka monitoring.

Kinesis Source #

Kinesis Sink #

HBase Connectors #

システム リソース #

システムリソースのレポートはデフォルトで無効です。When metrics.system-resource is enabled additional metrics listed below will be available on Job- and TaskManager. System resources metrics are updated periodically and they present average values for a configured interval (metrics.system-resource-probing-interval).

System resources reporting requires an optional dependency to be present on the classpath (for example placed in Flink’s lib directory):

• com.github.oshi:oshi-core:6.1.5 (licensed under MIT license)

Including it’s transitive dependencies:

• net.java.dev.jna:jna-platform:jar:5.10.0
• net.java.dev.jna:jna:jar:5.10.0

Failures in this regard will be reported as warning messages like NoClassDefFoundError logged by SystemResourcesMetricsInitializer during the startup.

システム CPU #

システム メモリ #

システム ネットワーク #

投機的実行 #

Metrics below can be used to measure the effectiveness of speculative execution.

End-to-End latency tracking #

Flinkによりシステム内を移動するレコードのレイテンシを追跡することができます。この機能はデフォルトで無効です。 To enable the latency tracking you must set the latencyTrackingInterval to a positive number in either the Flink configuration or ExecutionConfig.

At the latencyTrackingInterval, the sources will periodically emit a special record, called a LatencyMarker. マーカーはレコードがソースから発行された時間からのタイムスタンプを含みます。 Latency markers can not overtake regular user records, thus if records are queuing up in front of an operator, it will add to the latency tracked by the marker.

Note that the latency markers are not accounting for the time user records spend in operators as they are bypassing them. 特にマーカーは例えばウィンドウのバッファでレコードが費やした時間を計上しません。 Only if operators are not able to accept new records, thus they are queuing up, the latency measured using the markers will reflect that.

The LatencyMarkers are used to derive a distribution of the latency between the sources of the topology and each downstream operator. これらの分布はヒストグラム メトリクスとして報告されます。The granularity of these distributions can be controlled in the Flink configuration. For the highest granularity subtask Flink will derive the latency distribution between every source subtask and every downstream subtask, which results in quadratic (in the terms of the parallelism) number of histograms.

現在のところ、Flinkはクラスタ内の全てのマシーンのクロックが同期していることを仮定します。We recommend setting up an automated clock synchronisation service (like NTP) to avoid false latency results.

Warning Enabling latency metrics can significantly impact the performance of the cluster (in particular for subtask granularity). It is highly recommended to only use them for debugging purposes.

State access latency tracking #

Flink also allows to track the keyed state access latency for standard Flink state-backends or customized state backends which extending from AbstractStateBackend. この機能はデフォルトで無効です。 To enable this feature you must set the state.backend.latency-track.keyed-state-enabled to true in the Flink configuration.

Once tracking keyed state access latency is enabled, Flink will sample the state access latency every N access, in which N is defined by state.backend.latency-track.sample-interval. This configuration has a default value of 100. A smaller value will get more accurate results but have a higher performance impact since it is sampled more frequently.

As the type of this latency metrics is histogram, state.backend.latency-track.history-size will control the maximum number of recorded values in history, which has the default value of 128. A larger value of this configuration will require more memory, but will provide a more accurate result.

Warning Enabling state-access-latency metrics may impact the performance. It is recommended to only use them for debugging purposes.

REST API の統合 #

Metrics can be queried through the Monitoring REST API.

以下はサンプルのJSON応答を使った利用可能なエンドポイントのリストです。All endpoints are of the sample form http://hostname:8081/jobmanager/metrics, below we list only the path part of the URLs.

Values in angle brackets are variables, for example http://hostname:8081/jobs/<jobid>/metrics will have to be requested for example as http://hostname:8081/jobs/7684be6004e4e955c2a558a9bc463f65/metrics.

特定のエンティティのためのメトリクスのリクエスト:

• /jobmanager/metrics
• /taskmanagers/<taskmanagerid>/metrics
• /jobs/<jobid>/metrics
• /jobs/<jobid>/vertices/<vertexid>/subtasks/<subtaskindex>

それぞれの型の全てのエンティティを横断して集約されたメトリクスのリクエスト:

• /taskmanagers/metrics
• /jobs/metrics
• /jobs/<jobid>/vertices/<vertexid>/subtasks/metrics

それぞれの型の全てのエンティティの部分集合上の集約されたメトリクスのリクエスト:

• /taskmanagers/metrics?taskmanagers=A,B,C
• /jobs/metrics?jobs=D,E,F
• /jobs/<jobid>/vertices/<vertexid>/subtasks/metrics?subtask=1,2,3

Warning Metric names can contain special characters that you need to escape when querying metrics. For example, “a_+_b” would be escaped to “a_%2B_b”.

List of characters that should be escaped:

利用可能なメトリクスのリストのリクエスト:

GET /jobmanager/metrics

[
  {
    "id": "metric1"
  },
  {
    "id": "metric2"
  }
]

特定の(集約されていない)メトリクスの値のリクエスト:

GET taskmanagers/ABCDE/metrics?get=metric1,metric2

[
  {
    "id": "metric1",
    "value": "34"
  },
  {
    "id": "metric2",
    "value": "2"
  }
]

特定のメトリクスについての集約された値のリクエスト:

GET /taskmanagers/metrics?get=metric1,metric2

[
  {
    "id": "metric1",
    "min": 1,
    "max": 34,
    "avg": 15,
    "sum": 45
  },
  {
    "id": "metric2",
    "min": 2,
    "max": 14,
    "avg": 7,
    "sum": 16
  }
]

特定のメトリクスについての集約された値のリクエスト:

GET /taskmanagers/metrics?get=metric1,metric2&agg=min,max

[
  {
    "id": "metric1",
    "min": 1,
    "max": 34
  },
  {
    "id": "metric2",
    "min": 2,
    "max": 14
  }
]

ダッシュボードの統合 #

各タスクあるいはオペレータについて集められたメトリクスはダッシュボード内で可視化することもできます。On the main page for a job, select the Metrics tab. After selecting one of the tasks in the top graph you can select metrics to display using the Add Metric drop-down menu.

• Task metrics are listed as <subtask_index>.<metric_name>.
• Operator metrics are listed as <subtask_index>.<operator_name>.<metric_name>.

各メトリックはx軸が時間をy軸が計測された値を表す、個別のグラフとして可視化されるでしょう。 全てのグラフは自動的に10秒ごとに更新され、他のページに移動するまでそうし続けます。

可視化されるメトリクスの数には制限がありません; しかしながら数値メトリクスのみが可視化可能です。

Back to top