Flink Metrics監(jiān)控與 RestApi
本文轉(zhuǎn)載自微信公眾號(hào)「KK架構(gòu)師」，作者wangkai 。轉(zhuǎn)載本文請(qǐng)聯(lián)系KK架構(gòu)師公眾號(hào)。
一、Flink metrics簡介
Flink 的 metrics 是 Flink 公開的一個(gè)度量系統(tǒng)，metrics 也可以暴露給外部系統(tǒng)，通過在 Flink 配置文件 conf/flink-conf.yaml 配置即可，F(xiàn)link原生已經(jīng)支持了很多reporter，如 JMX、InfluxDB、Prometheus 等等。
我們也可以自定義指標(biāo)通過 metric 收集，實(shí)際開發(fā)時(shí)經(jīng)常需要查看當(dāng)前程序的運(yùn)行狀況，flink 提供了 UI 界面，有比較詳細(xì)的統(tǒng)計(jì)信息。
但是 UI 界面也有不完善的地方，比如想要獲取 flink 的實(shí)時(shí)吞吐。本文將詳細(xì)介紹如何通過 metric 監(jiān)控 flink 程序，自定義監(jiān)控指標(biāo)以及 metrics 在 flink 的 UI 界面的應(yīng)用。
二、Metrics在UI頁面上的應(yīng)用
在 flink 的 UI 的界面上我們點(diǎn)擊任務(wù)詳情，然后點(diǎn)擊 Task Metrics 會(huì)彈出如下的界面，在 add metic 按鈕上 我們可以添加我需要的監(jiān)控指標(biāo)。
注意：如果點(diǎn)擊 Task Metrics 沒有顯示 Add metics 點(diǎn)擊一下任務(wù)的 DAG 圖就會(huì)顯示出來，當(dāng)我們點(diǎn)擊了 DAG 圖中某個(gè)算子的名字，那么 Add metric 顯示的就是該算子的監(jiān)控指標(biāo)，且按照分區(qū)顯示，算子名前置的數(shù)字就是分區(qū)號(hào)。
三、各個(gè)指標(biāo)的含義
關(guān)于各個(gè)指標(biāo)的含義官網(wǎng)上有詳細(xì)介紹：
https://ci.apache.org/projects/flink/flink-docs-release-1.7/monitoring/metrics.html#availability
四、自定義監(jiān)控指標(biāo)
案例：在map算子內(nèi)計(jì)算輸入的總數(shù)據(jù)，設(shè)置 ：
- DataStream<String> userData = kafkaData.map(new RichMapFunction<String, String>() {
- Counter mapDataNub;
- @Override
- public void open(Configuration parameters) throws Exception {
- mapDataNub= getRuntimeContext()
- .getMetricGroup()
- .addGroup("flink_test_metric")
- .counter("mapDataNub");
- }
- @Override
- public String map(String s) {
- String s1 ="";
- try {
- String[] split = s.split(",");
- long userID = Long.parseLong(split[0]);
- long itemId = Long.parseLong(split[1]);
- long categoryId = Long.parseLong(split[2]);
- String behavior = split[3];
- long timestamp = Long.parseLong(split[4]);
- Map map = new HashMap();
- map.put("userID", userID);
- map.put("itemId", itemId);
- map.put("categoryId", categoryId);
- map.put("behavior", behavior);
- map.put("timestamp", timestamp);
- s1 = JSON.toJSONString(map);
- mapDataNub.inc();
- System.out.println("數(shù)據(jù)"+map.toString());
- } catch (NumberFormatException e) {
- e.printStackTrace();
- }
- return s1;
- }
程序啟動(dòng)之后就可以在任務(wù)的ui界面上查看
注意點(diǎn)：
搜索自定義或者查看某個(gè)指標(biāo)需要點(diǎn)擊DAG圖中對(duì)應(yīng)算子的名稱
指標(biāo)的前綴0，1，2....是指算子的分區(qū)數(shù)
進(jìn)行監(jiān)控時(shí)，盡量不要對(duì)算子進(jìn)行重命名，使用默認(rèn)的名字，這樣一套監(jiān)控程序可以監(jiān)控多個(gè)flink任務(wù)，比如對(duì)sink重新命名，如果不同的flink程序?qū)ink的命名不一樣，則一套監(jiān)控?zé)o法監(jiān)控多個(gè)flink程序
- .addSink(KafkaSink.getProducer()).name("kafka_sink");
五、Flink UI 不顯示算子數(shù)據(jù)接收和發(fā)送的條數(shù)
有時(shí)候我們Flink任務(wù)正常運(yùn)行,數(shù)據(jù)也可以打印,而且都保存到數(shù)據(jù)庫了,但是UI上面卻不顯示數(shù)據(jù)接收和發(fā)送的條數(shù) ，導(dǎo)致無法進(jìn)行指標(biāo)監(jiān)控和查查flink任務(wù)運(yùn)行的具體情況，這是什么原因?qū)е碌哪?
原因：是因?yàn)槟J(rèn)情況下Flink開啟了operator chain,所以當(dāng)flink程序所有的算子都在一個(gè)chain里面時(shí)，也就是在一個(gè)DAG(task)里面,所有沒有向下游發(fā)送數(shù)據(jù),所以顯示都為0。比如下圖的情況所有指標(biāo)都是0;
解決方案：第一種方法：在flink程序里添加自定義metric
第二種方法：使用startNewChain和disableChainin打斷程序默認(rèn)的operator chain
第三種方法：修改某個(gè)算子的并行度使其和上下游算子并行度不一致
六、Metric Reporter
Metrics可以暴露給外部系統(tǒng)，通過在flink配置文件conf/flink-conf.yaml配置即可，flink原生已經(jīng)支持了很多reporter，如JMX、InfluxDB、Prometheus等等，同時(shí)也支持自定義reporter。
Flink自帶了很多Reporter，包括JMX、InfluxDB、Prometheus等等，接下來介紹下InfluxDB Reporter的使用。
只需在flink配置文件conf/flink-conf.yaml中配置Influxdb相關(guān)信息即可，主要包括域名、端口號(hào)、用戶密碼等等。
flink1.10之后采用
- metrics.reporter.influxdb.factory.class: org.apache.flink.metrics.influxdb.InfluxdbReporterFactory
- metrics.reporter.influxdb.host: localhost
- metrics.reporter.influxdb.port: 8086
- metrics.reporter.influxdb.db: flink
- metrics.reporter.influxdb.consistency: ANY
- metrics.reporter.influxdb.connectTimeout: 60000
- metrics.reporter.influxdb.writeTimeout: 60000
- metrics.reporter.influxdb.interval: 30 SECONDS
flink1.10之前
- metrics.reporters: influxdb
- metrics.reporter.influxdb.class: org.apache.flink.metrics.influxdb.InfluxdbReporter
- metrics.reporter.influxdb.host: localhost
- metrics.reporter.influxdb.port: 8086
- metrics.reporter.influxdb.db: flink_monitor
- metrics.reporter.influxdb.username: flink-metrics
- metrics.reporter.influxdb.password: 123
注意事項(xiàng)：收集flinkSQL任務(wù)的監(jiān)控指標(biāo)，如果用戶書寫的sql語句 insert into 或者insert overwrite 中單引號(hào)帶有換行符，寫入influxdb會(huì)報(bào)錯(cuò)
查看influxdb收集到監(jiān)控信息，發(fā)現(xiàn)會(huì)自動(dòng)給我生成數(shù)據(jù)庫和measurement，所有的指標(biāo)都存儲(chǔ)在了具體的measurement中
七、flink metric監(jiān)控程序
前面介紹了flink公共的監(jiān)控指標(biāo)以及如何自定義監(jiān)控指標(biāo)，那么實(shí)際開發(fā)flink任務(wù)我們需要及時(shí)知道這些監(jiān)控指標(biāo)的數(shù)據(jù)，去獲取程序的健康值以及狀態(tài)。這時(shí)候就需要我們通過 flink REST API ，自己編寫監(jiān)控程序去獲取這些指標(biāo)。很簡單，當(dāng)我們知道每個(gè)指標(biāo)請(qǐng)求的URL,我們便可以編寫程序通過http請(qǐng)求獲取指標(biāo)的監(jiān)控?cái)?shù)據(jù)。
八、flink REST API監(jiān)控程序
為了獲取flink任務(wù)運(yùn)行狀態(tài)和吞吐量我們需要注意一下兩點(diǎn)：
- flink集群模式需要知道 JobManager 的地址和端口(5004)
- 對(duì)于 flink on yarn 模式來說，則需要知道 RM 代理的 JobManager UI 地址,例如 http://yarn-resource-manager-ui/proxy/application_155316436xxxx_xxxx
1.獲取flink任務(wù)運(yùn)行狀態(tài)(我們可以在瀏覽器進(jìn)行測試，輸入如下的連接)
http://yarn-resource-manager-ui/proxy/application_155316436xxxx_xxxx/jobs
返回的結(jié)果
- {
- jobs: [{
- id: "ce793f18efab10127f0626a37ff4b4d4",
- status: "RUNNING"
- }
- ]
- }
2.獲取 job 詳情
需要在/jobs/jobid
http://yarn-resource-manager-ui/proxy/application_155316436xxxx_xxxx/jobs/ce793f18efab10127f0626a37ff4b4d4
- {
- jid: "ce793f18efab10127f0626a37ff4b4d4",
- name: "Test",
- isStoppable: false,
- state: "RUNNING",
- start - time: 1551577191874,
- end - time: -1,
- duration: 295120489,
- now: 1551872312363,
- 。。。。。。
- 此處省略n行
- 。。。。。。
- }, {
- id: "cbc357ccb763df2852fee8c4fc7d55f2",
- parallelism: 12,
- operator: "",
- operator_strategy: "",
- description: "Source: Custom Source -> Flat Map",
- optimizer_properties: {}
- }
- ]
- }
- }
九、更靈活的方式獲取每個(gè)指標(biāo)的請(qǐng)求連接
有人可能會(huì)問，這么多指標(biāo)，難道我要把每個(gè)指標(biāo)的請(qǐng)求的URL格式都記住嗎?
今天教大家一個(gè)小技巧，一個(gè)前端技術(shù)，就是進(jìn)入flink任務(wù)的UI界面，按住F12進(jìn)入開發(fā)者模式，然后我們點(diǎn)擊任意一個(gè)metric指標(biāo),便能立即看到每個(gè)指標(biāo)的請(qǐng)求的URL。比如獲取flink任務(wù)的背壓情況：
如下圖我們點(diǎn)擊某一個(gè)task的status，按一下f12，便看到了backpressue,點(diǎn)開backpressue就是獲取任務(wù)背壓情況的連接如下：
http://127.0.0.1/proxy/application_12423523_133234/jobs/86eb310874aeccb37b58ae2892feced3/vertices/cbc357ccb763df2852fee8c4fc7d55f2/backpressure
請(qǐng)求連接返回的json字符串如下：我們可以獲取每一個(gè)分區(qū)的背壓情況，如果不是OK狀態(tài)便可以進(jìn)行任務(wù)報(bào)警，其他的指標(biāo)獲取監(jiān)控值都可以這樣獲取 簡單而又便捷。
十、案例：實(shí)時(shí)獲取yarn上flink任務(wù)運(yùn)行狀態(tài)
我們使用 flink REST API的方式，通過http請(qǐng)求實(shí)時(shí)獲取flink任務(wù)狀態(tài)，不是RUNNING狀態(tài)則進(jìn)行電話或郵件報(bào)警，達(dá)到實(shí)時(shí)監(jiān)控的效果。
- public class SendGet {
- public static String sendGet(String url) {
- String result = "";
- BufferedReader in = null;
- try {
- String urlNameString = url;
- URL realUrl = new URL(urlNameString);
- // 打開和URL之間的連接
- URLConnection connection = realUrl.openConnection();
- // 設(shè)置通用的請(qǐng)求屬性
- connection.setRequestProperty("accept", "*/*");
- connection.setRequestProperty("connection", "Keep-Alive");
- connection.setRequestProperty("user-agent",
- "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1;SV1)");
- // 建立實(shí)際的連接
- connection.connect();
- in = new BufferedReader(new InputStreamReader(
- connection.getInputStream()));
- String line;
- while ((line = in.readLine()) != null) {
- result += line;
- }
- } catch (Exception e) {
- System.out.println("發(fā)送GET請(qǐng)求出現(xiàn)異常！" + e);
- e.printStackTrace();
- }
- // 使用finally塊來關(guān)閉輸入流
- finally {
- try {
- if (in != null) {
- in.close();
- }
- } catch (Exception e2) {
- e2.printStackTrace();
- }
- }
- return result;
- }
- public static void main(String[] args) {
- String s = sendGet("http://127.0.0.1:5004/proxy/application_1231435364565_0350/jobs");
- JSONObject jsonObject = JSON.parseObject(s);
- String string = jsonObject.getString("jobs");
- String substring = string.substring(1, string.length() - 1);
- JSONObject jsonObject1 = JSONObject.parseObject(substring);
- String status = jsonObject1.getString("status");
- System.out.println(status);
- }
- }
結(jié)果
