Apache Flink 進階（八）：詳解 Metrics 原理與實戰(zhàn)

作者：劉彪
整理：毛鶴

本文由 Apache Flink Contributor 劉彪分享，本文對兩大問題進行了詳細的介紹，即什么是 Metrics、如何使用 Metrics，并對 Metrics 監(jiān)控實戰(zhàn)進行解釋說明。

什么是 Metrics？

Flink 提供的 Metrics 可以在 Flink 內(nèi)部收集一些指標，通過這些指標讓開發(fā)人員更好地理解作業(yè)或集群的狀態(tài)。由于集群運行后很難發(fā)現(xiàn)內(nèi)部的實際狀況，跑得慢或快，是否異常等，開發(fā)人員無法實時查看所有的 Task 日志，比如作業(yè)很大或者有很多作業(yè)的情況下，該如何處理？此時 Metrics 可以很好的幫助開發(fā)人員了解作業(yè)的當前狀況。

Metric Types

Metrics 的類型如下：

首先，常用的如 Counter，寫過 mapreduce 作業(yè)的開發(fā)人員就應該很熟悉 Counter，其實含義都是一樣的，就是對一個計數(shù)器進行累加，即對于多條數(shù)據(jù)和多兆數(shù)據(jù)一直往上加的過程。
第二，Gauge，Gauge 是最簡單的 Metrics，它反映一個值。比如要看現(xiàn)在 Java heap 內(nèi)存用了多少，就可以每次實時的暴露一個 Gauge，Gauge 當前的值就是heap使用的量。
第三，Meter，Meter 是指統(tǒng)計吞吐量和單位時間內(nèi)發(fā)生“事件”的次數(shù)。它相當于求一種速率，即事件次數(shù)除以使用的時間。
第四，Histogram，Histogram 比較復雜，也并不常用，Histogram 用于統(tǒng)計一些數(shù)據(jù)的分布，比如說 Quantile、Mean、StdDev、Max、Min 等。

Metric Group

Metric 在 Flink 內(nèi)部有多層結(jié)構(gòu)，以 Group 的方式組織，它并不是一個扁平化的結(jié)構(gòu)，Metric Group + Metric Name 是 Metrics 的唯一標識。

Metric Group 的層級有 TaskManagerMetricGroup 和TaskManagerJobMetricGroup，每個 Job 具體到某一個 task 的 group，task 又分為 TaskIOMetricGroup 和 OperatorMetricGroup。Operator 下面也有 IO 統(tǒng)計和一些 Metrics，整個層級大概如下圖所示。Metrics 不會影響系統(tǒng)，它處在不同的組中，并且 Flink支持自己去加 Group，可以有自己的層級。

?TaskManagerMetricGroup
    ?TaskManagerJobMetricGroup
        ?TaskMetricGroup
            ?TaskIOMetricGroup
            ?OperatorMetricGroup
                ?${User-defined Group} / ${User-defined Metrics}
                ?OperatorIOMetricGroup
?JobManagerMetricGroup
    ?JobManagerJobMetricGroup

JobManagerMetricGroup 相對簡單，相當于 Master，它的層級也相對較少。

Metrics 定義還是比較簡單的，即指標的信息可以自己收集，自己統(tǒng)計，在外部系統(tǒng)能夠看到 Metrics 的信息，并能夠?qū)ζ溥M行聚合計算。

如何使用 Metrics？

System Metrics

System Metrics，將整個集群的狀態(tài)已經(jīng)涵蓋得非常詳細。具體包括以下方面：

Master 級別和 Work 級別的 JVM 參數(shù)，如 load 和 time；其 Memory 劃分也很詳細，包括 heap 的使用情況，non-heap 的使用情況，direct 的使用情況，以及 mapped 的使用情況；Threads 可以看到具體有多少線程；還有非常實用的 Garbage Collection。
Network 使用比較廣泛，當需要解決一些性能問題的時候，Network 非常實用。Flink 不只是網(wǎng)絡傳輸，還是一個有向無環(huán)圖的結(jié)構(gòu)，可以看到它的每個上下游都是一種簡單的生產(chǎn)者消費者模型。Flink 通過網(wǎng)絡相當于標準的生產(chǎn)者和消費者中間通過有限長度的隊列模型。如果想要評估定位性能，中間隊列會迅速縮小問題的范圍，能夠很快的找到問題瓶頸。

?CPU
?Memory
?Threads
?Garbage Collection
?Network
?Classloader
?Cluster
?Availability
?Checkpointing
?StateBackend
?IO
?詳見: [https://ci.apache.org/projects/flink/flink-docs-release-1.8/monitoring/metrics.html#system-metrics](https://ci.apache.org/projects/flink/flink-docs-release-1.8/monitoring/metrics.html)

運維集群的人會比較關(guān)心 Cluster 的相關(guān)信息，如果作業(yè)太大，則需要非常關(guān)注 Checkpointing，它有可能會在一些常規(guī)的指標上無法體現(xiàn)出潛在問題。比如 Checkpointing 長時間沒有工作，數(shù)據(jù)流看起來沒有延遲，此時可能會出現(xiàn)作業(yè)一切正常的假象。另外，如果進行了一輪 failover 重啟之后，因為 Checkpointing 長時間沒有工作，有可能會回滾到很長一段時間之前的狀態(tài)，整個作業(yè)可能就直接廢掉了。
RocksDB 是生產(chǎn)環(huán)境當中比較常用的 state backend 實現(xiàn)，如果數(shù)據(jù)量足夠大，就需要多關(guān)注 RocksDB 的 Metrics，因為它隨著數(shù)據(jù)量的增大，性能可能會下降。

User-defined Metrics

除了系統(tǒng)的 Metrics 之外，F(xiàn)link 支持自定義 Metrics ，即 User-defined Metrics。上文說的都是系統(tǒng)框架方面，對于自己的業(yè)務邏輯也可以用 Metrics 來暴露一些指標，以便進行監(jiān)控。

User-defined Metrics 現(xiàn)在提及的都是 datastream 的 API，table、sql 可能需要 context 協(xié)助，但如果寫 UDF，它們其實是大同小異的。

Datastream 的 API 是繼承 RichFunction ，繼承 RichFunction 才可以有 Metrics 的接口。然后通過 RichFunction 會帶來一個 getRuntimeContext().getMetricGroup().addGroup(…) 的方法，這里就是 User-defined Metrics 的入口。通過這種方式，可以自定義 user-defined Metric Group。如果想定義具體的 Metrics，同樣需要用getRuntimeContext().getMetricGroup().counter/gauge/meter/histogram(…) 方法，它會有相應的構(gòu)造函數(shù)，可以定義到自己的 Metrics 類型中。

繼承 RichFunction
    ?Register user-defined Metric Group: getRuntimeContext().getMetricGroup().addGroup(…)
    ?Register user-defined Metric: getRuntimeContext().getMetricGroup().counter/gauge/meter/histogram(…)

User-defined Metrics Example

下面通過一段簡單的例子說明如何使用 Metrics。比如，定義了一個 Counter 傳一個 name，Counter 默認的類型是 single counter（Flink 內(nèi)置的一個實現(xiàn)），可以對 Counter 進行 inc（）操作，并在代碼里面直接獲取。

Meter 也是這樣，F(xiàn)link 有一個內(nèi)置的實現(xiàn)是 Meterview，因為 Meter 是多長時間內(nèi)發(fā)生事件的記錄，所以它是要有一個多長時間的窗口。平常用 Meter 時直接 markEvent()，相當于加一個事件不停地打點，最后用 getrate（）的方法直接把這一段時間發(fā)生的事件除一下給算出來。

Gauge 就比較簡單了，把當前的時間打出來，用 Lambda 表達式直接把 System::currentTimeMillis 打進去就可以，相當于每次調(diào)用的時候都會去真正調(diào)一下系統(tǒng)當天時間進行計算。

Histogram 稍微復雜一點，F(xiàn)link 中代碼提供了兩種實現(xiàn)，在此取一其中個實現(xiàn)，仍然需要一個窗口大小，更新的時候可以給它一個值。

這些 Metrics 一般都不是線程安全的。如果想要用多線程，就需要加同步，更多詳情請參考下面鏈接。

?Counter processedCount = getRuntimeContext().getMetricGroup().counter("processed_count");
  processedCount.inc();
?Meter processRate = getRuntimeContext().getMetricGroup().meter("rate", new MeterView(60));
  processRate.markEvent();
?getRuntimeContext().getMetricGroup().gauge("current_timestamp", System::currentTimeMillis);
?Histogram histogram = getRuntimeContext().getMetricGroup().histogram("histogram", new DescriptiveStatisticsHistogram(1000));
  histogram.update(1024);
?[https://ci.apache.org/projects/flink/flink-docs-release-1.8/monitoring/metrics.html#metric-types]

獲取 Metrics

獲取 Metrics 有三種方法，首先可以在 WebUI 上看到；其次可以通過 RESTful API 獲取，RESTful API 對程序比較友好，比如寫自動化腳本或程序，自動化運維和測試，通過 RESTful API 解析返回的 Json 格式對程序比較友好；最后，還可以通過 Metric Reporter 獲取，監(jiān)控主要使用 Metric Reporter 功能。

獲取 Metrics 的方式在物理架構(gòu)上是怎樣實現(xiàn)的？

了解背景和原理會對使用有更深刻的理解。WebUI 和 RESTful API 是通過中心化節(jié)點定期查詢把各個組件中的 Metrics 拉上來的實現(xiàn)方式。其中，fetch 不一定是實時更新的，默認為 10 秒，所以有可能在 WebUI 和 RESTful API 中刷新的數(shù)據(jù)不是實時想要得到的數(shù)據(jù)；此外，fetch 有可能不同步，比如兩個組件，一邊在加另一邊沒有動，可能是由于某種原因超時沒有拉過來，這樣是無法更新相關(guān)值的，它是 try best 的操作，所以有時我們看到的指標有可能會延遲，或許等待后相關(guān)值就更新了。

紅色的路徑通過 MetricFetcher，會有一個中心化的節(jié)點把它們聚合在一起展示。而 MetricReporter 不一樣，每一個單獨的點直接匯報，它沒有中心化節(jié)點幫助做聚合。如果想要聚合，需要在第三方系統(tǒng)中進行，比如常見的 TSDB 系統(tǒng)。當然，不是中心化結(jié)構(gòu)也是它的好處，它可以免去中心化節(jié)點帶來的問題，比如內(nèi)存放不下等，MetricReporter 把原始數(shù)據(jù)直接 Reporter 出來，用原始數(shù)據(jù)做處理會有更強大的功能。

Metric Reporter

Flink 內(nèi)置了很多 Reporter，對外部系統(tǒng)的技術(shù)選型可以參考，比如 JMX 是 java 自帶的技術(shù)，不嚴格屬于第三方。還有InfluxDB、Prometheus、Slf4j（直接打 log 里）等，調(diào)試時候很好用，可以直接看 logger，F(xiàn)link 本身自帶日志系統(tǒng)，會打到 Flink 框架包里面去。詳見：

?Flink 內(nèi)置了很多 Reporter，對外部系統(tǒng)的技術(shù)選型可以參考，詳見：[https://ci.apache.org/projects/flink/flink-docs-release-1.8/monitoring/metrics.html#reporter](https://ci.apache.org/projects/flink/flink-docs-release-1.8/monitoring/metrics.html)
?Metric Reporter Configuration Example
 metrics.reporters: your_monitor,jmx
 metrics.reporter.jmx.class: org.apache.flink.metrics.jmx.JMXReporter
 metrics.reporter.jmx.port: 1025-10000
 metrics.reporter.your_monitor.class: com.your_company.YourMonitorClass
 metrics.reporter.your_monitor.interval: 10 SECONDS
 metrics.reporter.your_monitor.config.a: your_a_value
 metrics.reporter.your_monitor.config.b: your_b_value

Metric Reporter 是如何配置的？如上所示，首先 Metrics Reporters 的名字用逗號分隔，然后通過 metrics.reporter.jmx.class 的 classname 反射找 reporter，還需要拿到 metrics.reporter.jmx.port 的配置，比如像第三方系統(tǒng)通過網(wǎng)絡發(fā)送的比較多。但要知道往哪里發(fā)，ip 地址、port 信息是比較常見的。此外還有 metrics.reporter.your_monitor.class 是必須要有的，可以自己定義間隔時間，F(xiàn)link 可以解析，不需要自行去讀，并且還可以寫自己的 config。

實戰(zhàn)：利用 Metrics 監(jiān)控

常用 Metrics 做自動化運維和性能分析。

自動化運維

自動化運維怎么做？

首先，收集一些關(guān)鍵的 Metrics 作為決策依據(jù)，利用 Metric Reporter 收集 Metrics 到存儲/分析系統(tǒng) (例如 TSDB)，或者直接通過 RESTful API 獲取。
有了數(shù)據(jù)之后，可以定制監(jiān)控規(guī)則，關(guān)注關(guān)鍵指標，F(xiàn)ailover、Checkpoint,、業(yè)務 Delay 信息。定制規(guī)則用途最廣的是可以用來報警，省去很多人工的工作，并且可以定制 failover 多少次時需要人為介入。
當出現(xiàn)問題時，有釘釘報警、郵件報警、短信報警、電話報警等通知工具。
自動化運維的優(yōu)勢是可以通過大盤、報表的形式清晰的查看數(shù)據(jù)，通過大盤時刻了解作業(yè)總體信息，通過報表分析優(yōu)化。

性能分析

性能分析一般遵循如下的流程：

首先從發(fā)現(xiàn)問題開始，如果有 Metrics 系統(tǒng)，再配上監(jiān)控報警，就可以很快定位問題。然后對問題進行剖析，大盤看問題會比較方便，通過具體的 System Metrics 分析，縮小范圍，驗證假設，找到瓶頸，進而分析原因，從業(yè)務邏輯、JVM、操作系統(tǒng)、State、數(shù)據(jù)分布等多維度進行分析；如果還不能找到問題原因，就只能借助 profiling 工具了。

實戰(zhàn)：“我的任務慢，怎么辦”

“任務慢，怎么辦？”可以稱之為無法解答的終極問題之一。

其原因在于這種問題是系統(tǒng)框架問題，比如看醫(yī)生時告訴醫(yī)生身體不舒服，然后就讓醫(yī)生下結(jié)論。而通常醫(yī)生需要通過一系列的檢查來縮小范圍，確定問題。同理，任務慢的問題也需要經(jīng)過多輪剖析才能得到明確的答案。

除了不熟悉 Flink 機制以外，大多數(shù)人的問題是對于整個系統(tǒng)跑起來是黑盒，根本不知道系統(tǒng)在如何運行，缺少信息，無法了解系統(tǒng)狀態(tài)。此時，一個有效的策略是求助 Metrics 來了解系統(tǒng)內(nèi)部的狀況，下面通過一些具體的例子來說明。

發(fā)現(xiàn)問題

比如下圖 failover 指標，線上有一個不是 0，其它都是 0，此時就發(fā)現(xiàn)問題了。

再比如下圖 Input 指標正常都在四、五百萬，突然跌成 0，這里也存在問題。

業(yè)務延時問題如下圖，比如處理到的數(shù)據(jù)跟當前時間比對，發(fā)現(xiàn)處理的數(shù)據(jù)是一小時前的數(shù)據(jù)，平時都是處理一秒之前的數(shù)據(jù)，這也是有問題的。

縮小范圍，定位瓶頸

當出現(xiàn)一個地方比較慢，但是不知道哪里慢時，如下圖紅色部分，OUT_Q 并發(fā)值已經(jīng)達到 100% 了，其它都還比較正常，甚至優(yōu)秀。到這里生產(chǎn)者消費者模型出現(xiàn)了問題，生產(chǎn)者 IN_Q 是滿的，消費者 OUT_Q 也是滿的，從圖中看出節(jié)點 4 已經(jīng)很慢了，節(jié)點 1 產(chǎn)生的數(shù)據(jù)節(jié)點 4 處理不過來，而節(jié)點 5 的性能都很正常，說明節(jié)點 1 和節(jié)點 4 之間的隊列已經(jīng)堵了，這樣我們就可以重點查看節(jié)點 1 和節(jié)點 4，縮小了問題范圍。

500 個 InBps 都具有 256 個 PARALLEL ，這么多個點不可能一一去看，因此需要在聚合時把 index 是第幾個并發(fā)做一個標簽。聚合按著標簽進行劃分，看哪一個并發(fā)是 100%。在圖中可以劃分出最高的兩個線，即線 324 和線 115，這樣就又進一步的縮小了范圍。

利用 Metrics 縮小范圍的方式如下圖所示，就是用 Checkpoint Alignment 進行對齊，進而縮小范圍，但這種方法用的較少。

多維度分析

分析任務有時候為什么特別慢呢？

當定位到某一個 Task 處理特別慢時，需要對慢的因素做出分析。分析任務慢的因素是有優(yōu)先級的，可以從上向下查，由業(yè)務方面向底層系統(tǒng)。因為大部分問題都出現(xiàn)在業(yè)務維度上，比如查看業(yè)務維度的影響可以有以下幾個方面，并發(fā)度是否合理、數(shù)據(jù)波峰波谷、數(shù)據(jù)傾斜；其次依次從 Garbage Collection、Checkpoint Alignment、State Backend 性能角度進行分析；最后從系統(tǒng)性能角度進行分析，比如 CPU、內(nèi)存、Swap、Disk IO、吞吐量、容量、Network IO、帶寬等。