本文我們通過官網來整體了解下flink 指標系統的系統性支持

?

本文主要關注：

• 如何注冊自定義指標，如何進行更新指標數據
• 指標定義的層級：即指標的scope
• 簡單介紹，指標如何報告給外部系統、有哪些系統指標
• 指標如何通過REST API獲取
• 在flink UI上創建Dashboard的方法

?

Flink exposes a metric system that allows gathering and exposing metrics to external systems.

flink 暴露了一個指標系統，可以收集和暴露指標給外部系統。

一. Registering metrics：向flink注冊新自己的metrics

1. 注冊metrics

任何繼承了RichFunction 的用戶函數，都可以通過調用：getRuntimeContext().getMetricGroup() ，來訪問flink的metric system。方法返回的MetricGroup可以用來創建和注冊新的指標。

?

2. Metric types:指標類型

flink支持 Counters, Gauges, Histograms and Meters.等四種指標類型。

2.1. Counter

計數器 (Counter) 用于計數某個指標。

• 可以使用 inc()/inc(long n) 或 dec()/dec(long n) 方法來增加或減少當前值。
• 可以通過在 MetricGroup 上調用 counter(String name) 來創建并注冊一個計數器。

public class MyMapper extends RichMapFunction<String, String> {private transient Counter counter;@Overridepublic void open(Configuration config) {this.counter = getRuntimeContext().getMetricGroup().counter("myCounter");}@Overridepublic String map(String value) throws Exception {this.counter.inc();return value;}
}

你也可以自己實現counter。

public class MyMapper extends RichMapFunction<String, String> {private transient Counter counter;@Overridepublic void open(Configuration config) {this.counter = getRuntimeContext().getMetricGroup().counter("myCustomCounter", new CustomCounter());}@Overridepublic String map(String value) throws Exception {this.counter.inc();return value;}
}

?

2.2. Gauge

可以提供任何數據類型，要使用Gauge你必須要實現Gauge接口，可以返回任何類型。

public class MyMapper extends RichMapFunction<String, String> {private transient int valueToExpose = 0;@Overridepublic void open(Configuration config) {getRuntimeContext().getMetricGroup().gauge("MyGauge", new Gauge<Integer>() {@Overridepublic Integer getValue() {return valueToExpose;}});}@Overridepublic String map(String value) throws Exception {valueToExpose++;return value;}
}

?

2.3. Histogram(ing)

直方圖（Histogram）用于測量長整型值的分布情況。

可以通過在 MetricGroup 上調用 histogram(String name, Histogram histogram) 來注冊一個直方圖。

public class MyMapper extends RichMapFunction<Long, Long> {private transient Histogram histogram;@Overridepublic void open(Configuration config) {this.histogram = getRuntimeContext().getMetricGroup().histogram("myHistogram", new MyHistogram());}@Overridepublic Long map(Long value) throws Exception {this.histogram.update(value);return value;}
}

ing

?

2.4. Meter

一個 Meter 用于測量平均吞吐量。

• 可以使用 markEvent() 方法注冊一個事件的發生。同時發生多個事件可以使用 markEvent(long n) 方法注冊。
• 在 MetricGroup 上調用 meter(String name, Meter meter) 來注冊一個 Meter。

?

二. Scope:指標作用域

每個度量指標都被分配了一個標識符和一組鍵值對，用于報告該度量指標。
這個標識符基于三個組件：在注冊度量指標時的用戶定義名稱，一個可選的用戶定義作用域，以及一個系統提供的作用域。

例如，如果 A.B 是系統作用域，C.D 是用戶作用域，E 是名稱，那么度量指標的標識符將是 A.B.C.D.E。

你可以通過在 Flink 配置文件中設置 metrics.scope.delimiter 鍵來配置標識符使用的分隔符（默認為 .）。

?

1. User Scope

你可以通過調用 MetricGroup#addGroup(String name)，MetricGroup#addGroup(int name)，或者 MetricGroup#addGroup(String key, String value) 來定義用戶作用域。

我們通過 MetricGroup#getMetricIdentifier 和 MetricGroup#getScopeComponents 方法返回的內容。

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

?

2. System Scope ing

?

3. User Variables

你可以通過調用 MetricGroup#addGroup(String key, String value) 來定義一個用戶變量。

這個方法會影響 MetricGroup#getMetricIdentifier、MetricGroup#getScopeComponents 和 MetricGroup#getAllVariables() 返回的內容。

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

?

三. Reporter ing

Flink 支持用戶將 Flink 的各項運行時指標發送給外部系統。

?

四. System metrics ing

默認情況下，Flink會收集多個度量指標，這些指標能夠深入了解當前的狀態。

?

五. REST API integration

度量指標可以通過監控REST API查詢。以下是可用端點列表及其示例JSON響應。

序號	metric類型	API
1	特定實體的metric	- /jobmanager/metrics - /taskmanagers/<taskmanagerid>/metrics - /jobs/<jobid>/metrics - /jobs/<jobid>/vertices/<vertexid>/subtasks/<subtaskindex>
2	實體的聚合metric	- /taskmanagers/metrics - /jobs/metrics - /jobs/<jobid>/vertices/<vertexid>/subtasks/metrics - /jobs/<jobid>/vertices/<vertexid>/jm-operator-metrics
3	實體子集上聚合的metric	- /taskmanagers/metrics?taskmanagers=A,B,C - /jobs/metrics?jobs=D,E,F - /jobs/<jobid>/vertices/<vertexid>/subtasks/metrics?subtask=1,2,3

?

六. Dashboard integration

可以在儀表板中可視化每個任務或操作符收集的度量指標。在作業的主頁面上，選擇“Metrics”選項卡。在頂部圖表中選擇一個任務后，您可以使用“添加度量指標”下拉菜單選擇要顯示的度量指標。如下圖：

• 任務度量指標列出為<子任務索引>.<度量名稱>。
• 操作符度量指標列出為 <子任務索引>.<操作符名稱>.<度量名稱>。

• 每個度量指標將顯示為單獨的圖表，其中 x 軸代表時間，y 軸表示測量值。
• 所有圖表每隔10秒自動更新一次，在導航到其他頁面時仍會繼續更新。
• 可視化的度量指標數量沒有限制，但是只有數值型度量指標可以被可視化顯示。

?