Metrics在Flink系統中的使用分析

  • 2020 年 1 月 27 日
  • 筆記

什麼是metrics：

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

Metrics 的類型如下：

1、常用的如 Counter，寫過 mapreduce 作業的開發人員就應該很熟悉 Counter，其實含義都是一樣的，就是對一個計數器進行累加，即對於多條數據和多兆數據一直往上加的過程。 2、Gauge，Gauge 是最簡單的 Metrics，它反映一個值。比如要看現在 Java heap 內存用了多少，就可以每次實時的暴露一個 Gauge，Gauge 當前的值就是heap使用的量。 3、Meter，Meter 是指統計吞吐量和單位時間內發生「事件」的次數。它相當於求一種速率，即事件次數除以使用的時間。 4、Histogram，Histogram 比較複雜，也並不常用，Histogram 用於統計一些數據的分佈，比如說 Quantile、Mean、StdDev、Max、Min 等。

Metric Group

Metric 在 Flink 內部有多層結構，以 Group 的方式組織，它並不是一個扁平化的結構，Metric Group + Metric Name 是 Metrics 的唯一標識。

Metric Group 的層級有 TaskManagerMetricGroup 和TaskManagerJobMetricGroup，每個 Job 具體到某一個 task 的 group，task 又分為 TaskIOMetricGroup 和 OperatorMetricGroup。

Operator 下面也有 IO 統計和一些 Metrics，整個層級大概如下圖所示。Metrics 不會影響系統，它處在不同的組中，並且 Flink支持自己去加 Group，可以有自己的層級。

•TaskManagerMetricGroup

  •TaskManagerJobMetricGroup

    •TaskMetricGroup

      •TaskIOMetricGroup

      •OperatorMetricGroup

        •${User-defined Group} / ${User-defined Metrics}

        •OperatorIOMetricGroup

•JobManagerMetricGroup

  •JobManagerJobMetricGroup

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

Metrics 定義還是比較簡單的，即指標的信息可以自己收集，自己統計，在外部系統能夠看到 Metrics 的信息，並能夠對其進行聚合計算。

如何使用 Metrics？

System Metrics

System Metrics，將整個集群的狀態已經涵蓋得非常詳細。具體包括以下方面：

Master 級別和 Work 級別的 JVM 參數，如 load 和 time；其 Memory 劃分也很詳細，包括 heap 的使用情況、non-heap 的使用情況、direct 的使用情況，以及 mapped 的使用情況；Threads 可以看到具體有多少線程；還有非常實用的 Garbage Collection。 Network 使用比較廣泛，當需要解決一些性能問題的時候，Network 非常實用。Flink 不只是網絡傳輸，還是一個有向無環圖的結構，可以看到它的每個上下游都是一種簡單的生產者消費者模型。Flink 通過網絡相當於標準的生產者和消費者中間通過有限長度的隊列模型。如果想要評估定位性能，中間隊列會迅速縮小問題的範圍，能夠很快的找到問題瓶頸。 •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) 運維集群的人會比較關心 Cluster 的相關信息，如果作業太大，則需要非常關注 Checkpointing，它有可能會在一些常規的指標上無法體現出潛在問題。比如 Checkpointing 長時間沒有工作，數據流看起來沒有延遲，此時可能會出現作業一切正常的假象。另外，如果進行了一輪 failover 重啟之後，因為 Checkpointing 長時間沒有工作，有可能會回滾到很長一段時間之前的狀態，整個作業可能就直接廢掉了。 RocksDB 是生產環境當中比較常用的 state backend 實現，如果數據量足夠大，就需要多關注 RocksDB 的 Metrics，因為它隨着數據量的增大，性能可能會下降。

自定義Metrics：

除了系統的 Metrics 之外，Flink 支持自定義 Metrics ，即 User-defined Metrics。上文說的都是系統框架方面，對於自己的業務邏輯也可以用 Metrics 來暴露一些指標，以便進行監控。

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

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

使用代碼示例：

Counter： 用與存儲數值類型，比如統計數據輸入、輸出總數量。

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;    }  }

Gauge: 可以用來存儲任何類型，前提要實現org.apache.flink.metrics.Gauge接口，重寫getValue方法，如果返回類型為Object則該類需要重寫toString方法。

有些場景下，需要根據業務計算出指標，則Gauge使用起來更靈活。

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;    }  }

項目實現代碼舉例：

添加自定義監控指標，以flink1.5的Kafka讀取以及寫入為例，添加rps、dirtyData等相關指標信息。�kafka讀取和寫入重點是先拿到RuntimeContex初始化指標，並傳遞給要使用的序列類，通過重寫序列化和反序列化方法，來更新指標信息。
不加指標的kafka數據讀取、寫入Demo。
public class FlinkEtlTest {      private static final Logger logger = LoggerFactory.getLogger(FlinkEtlTest.class);        public static void main(String[] args) throws Exception {          final ParameterTool params = ParameterTool.fromArgs(args);          String jobName = params.get("jobName");            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();            /** 設置kafka數據 */          String topic = "myTest01";          Properties props = new Properties();          props.setProperty("bootstrap.servers", "localhost:9092");          props.setProperty("zookeeper.quorum", "localhost:2181/kafka");            // 使用FlinkKafkaConsumer09以及SimpleStringSchema序列化類，讀取kafka數據          FlinkKafkaConsumer09<String> consumer09 = new FlinkKafkaConsumer09(topic, new SimpleStringSchema(), props);          consumer09.setStartFromEarliest();            // 使用FlinkKafkaProducer09和SimpleStringSchema反序列化類，將數據寫入kafka          String sinkBrokers = "localhost:9092";          FlinkKafkaProducer09<String> myProducer = new FlinkKafkaProducer09<>(sinkBrokers, "myTest01", new SimpleStringSchema());              DataStream<String> kafkaDataStream = env.addSource(consumer09);          kafkaDataStream = kafkaDataStream.map(str -> {              logger.info("map receive {}",str);              return str.toUpperCase();          });            kafkaDataStream.addSink(myProducer);            env.execute(jobName);      }      }

下面重新複寫flink的

FlinkKafkaConsumer09
FlinkKafkaProducer09

方法，加入metrics的監控。

為kafka讀取添加相關指標
  • 繼承FlinkKafkaConsumer09，獲取它的RuntimeContext，使用當前MetricGroup初始化指標參數。
public class CustomerFlinkKafkaConsumer09<T> extends FlinkKafkaConsumer09<T> {        CustomerSimpleStringSchema customerSimpleStringSchema;      // 構造方法有多個      public CustomerFlinkKafkaConsumer09(String topic, DeserializationSchema valueDeserializer, Properties props) {          super(topic, valueDeserializer, props);          this.customerSimpleStringSchema = (CustomerSimpleStringSchema) valueDeserializer;      }        @Override      public void run(SourceContext sourceContext) throws Exception {          //將RuntimeContext傳遞給customerSimpleStringSchema          customerSimpleStringSchema.setRuntimeContext(getRuntimeContext());         // 初始化指標          customerSimpleStringSchema.initMetric();          super.run(sourceContext);      }  }

重寫SimpleStringSchema類的反序列化方法，當數據流入時變更指標。

public class CustomerSimpleStringSchema extends SimpleStringSchema {        private static final Logger logger = LoggerFactory.getLogger(CustomerSimpleStringSchema.class);        public static final String DT_NUM_RECORDS_RESOVED_IN_COUNTER = "dtNumRecordsInResolve";      public static final String DT_NUM_RECORDS_RESOVED_IN_RATE = "dtNumRecordsInResolveRate";      public static final String DT_DIRTY_DATA_COUNTER = "dtDirtyData";      public static final String DT_NUM_BYTES_IN_COUNTER = "dtNumBytesIn";      public static final String DT_NUM_RECORDS_IN_RATE = "dtNumRecordsInRate";        public static final String DT_NUM_BYTES_IN_RATE = "dtNumBytesInRate";      public static final String DT_NUM_RECORDS_IN_COUNTER = "dtNumRecordsIn";            protected transient Counter numInResolveRecord;      //source RPS      protected transient Meter numInResolveRate;      //source dirty data      protected transient Counter dirtyDataCounter;        // tps      protected transient Meter numInRate;      protected transient Counter numInRecord;        //bps      protected transient Counter numInBytes;      protected transient Meter numInBytesRate;            private transient RuntimeContext runtimeContext;        public void initMetric() {          numInResolveRecord = runtimeContext.getMetricGroup().counter(DT_NUM_RECORDS_RESOVED_IN_COUNTER);          numInResolveRate = runtimeContext.getMetricGroup().meter(DT_NUM_RECORDS_RESOVED_IN_RATE, new MeterView(numInResolveRecord, 20));          dirtyDataCounter = runtimeContext.getMetricGroup().counter(DT_DIRTY_DATA_COUNTER);            numInBytes = runtimeContext.getMetricGroup().counter(DT_NUM_BYTES_IN_COUNTER);          numInRecord = runtimeContext.getMetricGroup().counter(DT_NUM_RECORDS_IN_COUNTER);            numInRate = runtimeContext.getMetricGroup().meter(DT_NUM_RECORDS_IN_RATE, new MeterView(numInRecord, 20));          numInBytesRate = runtimeContext.getMetricGroup().meter(DT_NUM_BYTES_IN_RATE , new MeterView(numInBytes, 20));            }      // 源表讀取重寫deserialize方法      @Override      public String deserialize(byte[] value) {          // 指標進行變更          numInBytes.inc(value.length);          numInResolveRecord.inc();          numInRecord.inc();          try {              return super.deserialize(value);          } catch (Exception e) {              dirtyDataCounter.inc();          }          return "";      }          public void setRuntimeContext(RuntimeContext runtimeContext) {          this.runtimeContext = runtimeContext;      }  }

代碼中使用自定義的消費者進行調用：

CustomerFlinkKafkaConsumer09<String> consumer09 = new CustomerFlinkKafkaConsumer09(topic, new CustomerSimpleStringSchema(), props);
為kafka寫入添加相關指標
  • 繼承FlinkKafkaProducer09類，重寫open方法，拿到RuntimeContext，初始化指標信息傳遞給CustomerSinkStringSchema。
public class  CustomerFlinkKafkaProducer09<T> extends FlinkKafkaProducer09<T> {        public static final String DT_NUM_RECORDS_OUT = "dtNumRecordsOut";      public static final String DT_NUM_RECORDS_OUT_RATE = "dtNumRecordsOutRate";        CustomerSinkStringSchema schema;        public CustomerFlinkKafkaProducer09(String brokerList, String topicId, SerializationSchema serializationSchema) {          super(brokerList, topicId, serializationSchema);          this.schema = (CustomerSinkStringSchema) serializationSchema;      }            @Override      public void open(Configuration configuration) {          producer = getKafkaProducer(this.producerConfig);            RuntimeContext ctx = getRuntimeContext();          Counter counter = ctx.getMetricGroup().counter(DT_NUM_RECORDS_OUT);          //Sink的RPS計算          MeterView meter = ctx.getMetricGroup().meter(DT_NUM_RECORDS_OUT_RATE, new MeterView(counter, 20));          // 將counter傳遞給CustomerSinkStringSchema          schema.setCounter(counter);            super.open(configuration);      }    }

重寫SimpleStringSchema的序列化方法 

public class CustomerSinkStringSchema extends SimpleStringSchema {        private static final Logger logger = LoggerFactory.getLogger(CustomerSinkStringSchema.class);        private Counter sinkCounter;        @Override      public byte[] serialize(String element) {          logger.info("sink data {}", element);          sinkCounter.inc();          return super.serialize(element);  //複寫serialize方法，序列化繼續使用父類提供的序列化方法      }        public void setCounter(Counter counter) {          this.sinkCounter = counter;      }  }

新的kafkaSinkApi使用

CustomerFlinkKafkaProducer09<String> myProducer = new CustomerFlinkKafkaProducer09<>(sinkBrokers, "mqTest01", new CustomerSinkStringSchema());

獲取 Metrics

這樣就可以在監控框架裏面看到採集的指標信息了，

比如flink_taskmanager_job_task_operator_dtDirtyData指標，dtDirtyData是自己添加的指標，前面的字符串是operator默認使用的metricGroup。

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

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

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

VirMach 便宜 VPS

QNews

QNews