八种Flink任务监控告警方式

目录

一、Flink应用分析

1.1 Flink任务生命周期

1.2 Flink应用告警视角分析

二、监控告警方案说明

2.1 监控消息队中间件消费者偏移量

2.2 通过调度系统监控Flink任务运行状态

2.3 引入开源服的SDK工具实现

2.4 调用FlinkRestApi实现任务监控告警

2.5 定时去查询目标库最大时间和当前时间做对比

2.6 自定义指标Reporter的SDK

2.7 任务日志告警

2.8 运行任务探活

三、总结

前言:Flink作为一个高性能实时计算引擎，可灵活的嵌入各种场景，许多团队为了实现业务交付，选择了Flink作为解决方案；但是随着Flink应用的增多且出现线上事故，对Flink任务异常的监控告警成为迫切需求；但是如何实现Flink任务异常监控告警，成为了新的问题；本文将从多个角度讲述Flink任务监控告警实现方案。

一、Flink应用分析

       告警可以从多个角度实现；我们先分析Flink任务运行的生命周期，然后拆解每个部分，分析可以从那些角度去监控Flink任务的异常。

1.1 Flink任务生命周期

按读取数据源：有如Kafka、RocketMq、Pulsar等消息队列，还有其他数据源；区别在是否有记录消费者信息的数据标识；

Flink的运行模式：session、per-job、application；三类运行模式可以分为两类场景：单独运行的任务（per和application），还有Flink集群统一提供资源运行的任务（session）；

任务场景：离线任务还是实时任务；

Flink任务应用结构图如下：

1.2 Flink应用告警视角分析

从数据源头：

1.对于消息队列这种，本身拥有记录消费者偏移量概念的中间件，可以通过监控消费者偏移量的变化来监控Flink任务运行的异常情况；

从任务运行时：

2.任务层可以通过调度系统的告警插件，监控任务运行结果和任务运行状态而监控任务；

3.也可以在Flink任务内部引入开源SDK配置开源工具实现；

4.或者调用FlinkRestApi实现任务监控告警；

从输出结果上：

5.可以定时去查询输出结果最后的时间；

6.或者在Flink任务里引入Flink的指标SDK，自定义Flink任务的指标采集，将结果测流输出到目标端，自定义监控告警和分析；

其他的方式：

7.日志告警，捕捉运行日志，通过关键词监控告警；

8.运行任务定时探活；

二、监控告警方案说明

       钉钉、微信、邮件、电话、http等属于告警方式的选择，这里侧重讲对于运行异常事件信息的捕捉。

2.1 监控消息队中间件消费者偏移量

      类似Kafka或者RocketMQ这类拥有记录消费者消费队列信息的中间件，可以通过服务自身的RestAPI，定时计算消费者消费数据lag条数；

以下是Kafka消费者告警配置页面：

这需要后端自定义实现;

实现方式如下：定时通过调用Kafka自己提供的RestApi将Topic和各消费者同步到Mysql,然后配置要监控Topic的消费者告警阈值和告警人员，每隔一分钟定时计算该消费者的lag，如果Flink任务出现异常，本身不提交offset了，数据积压量大于阈值就告警。

2.2 通过调度系统监控Flink任务运行状态

市场上有一些任务调度系统，比如dolphinscheduler、StreamX等，除了提供任务发布的能力，还自带监控告警功能，通过使用这类产品，也能做到监控告警能力。

比如dolphinscheduler：

Flink任务发布功能:

告警功能插件：

比如StreamPark:

Flink任务发布能力：

告警功能插件：

2.3 引入开源服务的SDK工具实现

       博客上对于Flink监控告警推荐最多的一种方式就是，prometheus + pushgateway + grafana这套方案；这套方案需要安装维护prometheus和grafana这两个产品，比较重，但是这套方案除了可以做到任务监控，还可以做到任务指标级的分析，这对于后续的任务性能优化有比较好的支持。

具体操作步骤如下：

1.安装好prometheus + pushgateway这两个服务；

2.在Flink代码里加入以下依赖：

  <!-- Prometheus Metrics Reporter --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-metrics-prometheus</artifactId><version>${flink-version}</version></dependency>

3.在部署Flink的配置文件里

将flink-metrics-prometheus-1.14.3.jar 包放入到flink安装目录/lib下

修改flink-conf.yaml配置文件，设置属性如下：

Example configuration: metrics.reporter.promgateway.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter metrics.reporter.promgateway.host: localhost metrics.reporter.promgateway.port: 9091 metrics.reporter.promgateway.jobName: myJob metrics.reporter.promgateway.randomJobNameSuffix: true metrics.reporter.promgateway.deleteOnShutdown: false metrics.reporter.promgateway.groupingKey: k1=v1;k2=v2 metrics.reporter.promgateway.interval: 60 SECONDS

然后启动运行任务，指标数据就自动推送到pushgateway里了，prometheus会从CC里拉取数据到自己的服务里，如下：

在grafana里导入prometheus源，配置指标就可以看到各种指标的运行状态：

总结：这种方案需要四个步骤：

1.启动prometheus+pushgateway+grafana服务；

2.配置Flink安装目录的配置文件、导入prometheus的lib包；

3.然后在Flink任务里引入一个prometheus的SDK，一起打包启动，指标就可以在prometheus看到；

4.通过grafana做分析看板和配置告警规则，驱动事件告警；

       这种方式都是开源服务功能，但是需要维护和理解成本，对于一些轻业务团队有负担，但是对有很多Flink任务的团队，这是一种可用的方案，后续还可以基于历史指标分析，做到内存级的性能优化；

2.4 调用FlinkRestApi实现任务监控告警

       这里要搞清楚Flink集群的生命和Flink任务的生命周期这两个概念；Flink集群按生命周期来分，运行方式可以分为session模式和其他模式两种；这两种的区别分别是，Flink集群和Flink任务的资源是否一起释放；这关系到是否可以稳定的通过FlinkRestApi捕捉到任务运行状态；

       对于Flink Sesion集群，Flink任务可以反复提交，集群的URL是不会变的，可以通过固定的URL监控到Flink任务的运行状态；

       对于per-job和application运行方式，Flink任务web的URL是不固定的，需要每次都捕捉到启动时的Url才能通过url调用RestAPI返回查询指标；

sesion集群样式：

per-job和application运行模式提交的任务，只会有一个任务，且url是随机的。

2.5 定时去查询目标库最大时间和当前时间做对比

       这种方式是公司的DB团队给我的想法，并且他们最初也是这么做的，虽然操作上不美观，无法大面积，且性能上会造成一些影响，但确实可以轻量级的实现对任务异常的监控；

具体是怎么做的呢？

       对于实时任务，数据都是实时捕捉的，写入目标库的时候，数据带有当前时间字段，业务理想状态下，数据会一直产生，查询目标库时间最大的数据与当前时间匹配，超出阈值时间范围就告警；不理想状态，将特殊时间段监控去掉就行；这种方式在生成业务种确实能满足任务的异常监控告警需求。

     要查询最大时间的数据，可以使用如下的 SQL 语句：

 SELECT time_column FROM table_name ORDER BY time_column DESC LIMIT 1;

2.6 自定义指标Reporter的SDK

1.引入Flink自带的指标SDK：

<!-- Prometheus Metrics Reporter --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-metrics-prometheus</artifactId><version>${flink-version}</version></dependency>

2.类似prometheus，将指标类的一些参数，自定义捕捉写到目标库（将推送到pushgateway改成推送到Kafka），然后通过目标库的数据自己做任务异常监控分析；

       这种方式就是避免了开源维护的成本，可以使用产品线自研的一套UI和采集中间件做数据管理，减轻了维护成本。

大致步骤是：

1.自定义 ReporterFactory 实现 MetricReporterFactory 接口中的 createMetricReporter 方法。

2.自定义 Reporter 继承 AbstractReporter 实现 Scheduled 接口中的相关方法

3.在 META-INF/services 下的配置文件中添加对应的实现类，然后在Flink配置里自定义参数。

以写入Kafka为例：

实现KafkaReporterFactory：

package org.apache.flink.metrics.kafka;import org.apache.flink.metrics.reporter.MetricReporter;
import org.apache.flink.metrics.reporter.MetricReporterFactory;import java.util.Properties;/*** @Description:* @author:i7Yang* @create 2024-01-26 20:19**/
public class KafkaReporterFactory implements MetricReporterFactory {@Overridepublic MetricReporter createMetricReporter(Properties properties) {return new KafkaReporter();}
}

实现自定义KafkaReporter：

package org.apache.flink.metrics.kafka;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.metrics.Metric;
import org.apache.flink.metrics.MetricConfig;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.reporter.AbstractReporter;
import org.apache.flink.metrics.reporter.MetricReporter;
import org.apache.flink.metrics.reporter.Scheduled;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.util.*;
import java.util.stream.Collectors;/*** {@link MetricReporter} that exports {@link Metric Metrics} via Kafka.*/public class KafkaReporter extends AbstractReporter implements Scheduled {private static final Logger LOGGER = LoggerFactory.getLogger(KafkaReporter.class);static final String JOB_ID_VARIABLE = "<job_id>";static final String JOB_NAME_VARIABLE = "<job_name>";private KafkaProducer<String, String> kafkaProducer;private List<String> metricsFilter = new ArrayList<>();private String topic;private String jobName;private String jobId;@Overridepublic void open(MetricConfig metricConfig) {String bootstrapServer = metricConfig.getString("bootstrapServers", "master:9092,storm1:9092,storm2:9092");String filter = metricConfig.getString("filter", "");String chunkSize = metricConfig.getString("chunkSize", "5");String topic = metricConfig.getString("topic", "flink_metric");Properties properties = new Properties();properties.setProperty("bootstrap.servers", bootstrapServer);properties.setProperty("acks", "all");properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");ClassLoader classLoader = Thread.currentThread().getContextClassLoader();Thread.currentThread().setContextClassLoader(null);kafkaProducer = new KafkaProducer<>(properties);Thread.currentThread().setContextClassLoader(classLoader);if (StringUtils.isNotEmpty(filter)) {this.metricsFilter.addAll(Arrays.asList(filter.split(",")));}this.chunkSize = Integer.parseInt(chunkSize);this.topic = topic;// 获取任务的 jobNamethis.jobName = metricConfig.getString("FLINK_JOB_NAME", null);LOGGER.info("job name: {}", jobName);}@Overridepublic void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {Map<String, String> allVariables = group.getAllVariables();String jobID = allVariables.get(JOB_ID_VARIABLE);if (jobID != null && this.jobId == null) {this.jobId = jobID;}String jobName = allVariables.get(JOB_NAME_VARIABLE);if (jobName != null && this.jobName == null) {this.jobName = jobName;}LOGGER.info("job id: {}, job name: {}", this.jobId, this.jobName);LOGGER.info("metric group name: {}, metric name: {}", group.getAllVariables(), metricName);// 只有在 filter 里面的 metric 才会被添加super.notifyOfAddedMetric(metric, metricName, group);}@Overridepublic void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group) {super.notifyOfRemovedMetric(metric, metricName, group);}@Overridepublic void close() {if (kafkaProducer != null) {kafkaProducer.close();}}@Overridepublic void report() {synchronized (this) {tryReport();}}private void tryReport() {Map<String, Object> metricMap = new HashMap<>();metricMap.put("jobId", this.jobId);metricMap.put("jobName", this.jobName);JSONArray jsonArray = new JSONArray();gauges.forEach((gauge, metricName) -> {JSONObject jsonObject = new JSONObject();jsonObject.put("metricName", metricName);jsonObject.put("value", gauge.getValue());jsonObject.put("type", "Gauge");jsonArray.add(jsonObject);});counters.forEach((counter, metricName) -> {JSONObject jsonObject = new JSONObject();jsonObject.put("metricName", metricName);jsonObject.put("value", counter.getCount());jsonObject.put("type", "Counter");jsonArray.add(jsonObject);});histograms.forEach((histogram, metricName) -> {JSONObject jsonObject = new JSONObject();jsonObject.put("metricName", metricName);jsonObject.put("value", histogram.getCount());jsonObject.put("type", "Histogram");jsonArray.add(jsonObject);});meters.forEach((meter, metricName) -> {JSONObject jsonObject = new JSONObject();jsonObject.put("metricName", metricName);jsonObject.put("value", meter.getCount());jsonObject.put("type", "Meter");jsonArray.add(jsonObject);});metricMap.put("metrics", jsonArray);ProducerRecord<String, String> record = new ProducerRecord<>(this.topic, this.jobId, JSONObject.toJSONString(metricMap));kafkaProducer.send(record);}@Overridepublic String filterCharacters(String input) {return input;}
}

 flink 的配置文件中设置一下 kafka reporter：

metrics.reporter.kafka.factory.class: org.apache.flink.metrics.kafka.KafkaReporterFactory
metrics.reporter.kafka.bootstrapServers: master:9092,storm1:9092,storm2:9092
metrics.reporter.kafka.topic: flink_metric
metrics.reporter.kafka.filter: inPoolUsage,outPoolUsage,numberOfCompletedCheckpoints,lastCheckpointFullSize,numBytesOutPerSecond,numBuffersOutPerSecond,numRecordsInPerSecond
metrics.reporter.kafka.interval: 20 SECONDS

2.7 任务日志告警

       将Flink的运行任务集中采集，文件日志用LogStagsh,指标日志可在应用里埋点，然后通过日志做告警管理。

2.8 运行任务探活

      上面2.4节讲了Flink的sesion运行模式，可以通过FlinkRestApi获取运行状态和指标；但是对于per-job和applicaiton运行方式，任务异常失败后，restApi是不存在，但是对于其使用的资源管理器，可以捕捉到任务运行状态；比如yarn，可以通过shell查询到任务的存活情况，可以定时去探活或获取url获取运行时指标。

使用yarn做Flink任务资源管理的命令：

定时监控flink任务状态：

yarn application -list | grep -w flink任务名 字 | awk '{print $1}'

返回flink任务url链接：

yarn application -list | grep -w flink 任务名字 | awk '{print $10}'

三、总结

       Flink任务告警方式的选择，要从任务的使用情况和期盼来考量；简单的使用，且任务少，可以用监控目标数据库的数据写入情况、per-job和application运行任务探活、Sesion运行方式通过RestApi来告警；特定场景的业务可以靠监控存储中间偏移量来告警；通用大规模应用场景可以通过采集运行时日志、使用调度平台，使用调度平台、引入开源SDK方式、自定义SDK写入通用系统通用系统里方式选择。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/661089.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【数据结构（C语言）】树、二叉树详解

【数据结构（C语言）】树、二叉树详解

目录 文章目录 前言 一、树的概念及结构 1.1 树的概念 1.2 树的相关概念 1.3 树的表示 1.4 树在实际中的运用 二、二叉树的概念及结构 2.1 二叉树的概念 2.2 二叉树的基本形态 ​编辑2.3 特殊的二叉树 2.4 二叉树的性质 2.5 二叉树的存储结构 三、二叉树的顺序结…
阅读更多...
堆溢出：深入理解与防范

堆溢出：深入理解与防范

在计算机安全领域中&#xff0c;堆溢出是一种常见的安全漏洞&#xff0c;它涉及到程序中的内存分配和管理。本文将深入探讨堆溢出的概念、原因以及防范措施。 首先&#xff0c;我们需要了解什么是堆。在许多编程语言中&#xff0c;如C和C&#xff0c;堆是用于动态内存分配的区…
阅读更多...
adb push 将电脑中的文件传输到安卓开发板

adb push 将电脑中的文件传输到安卓开发板

1. adb remount 重新挂载设备的文件系统&#xff0c;以便可以对设备进行读写操作&#xff0c;通常情况下&#xff0c;安卓开发板在连接到计算机后&#xff0c;设备的文件系统会被挂载为只读文件系统&#xff0c;重新挂载后变成可读可写权限 C:\Users\Administrator>adb re…
阅读更多...
高等数学：积分

高等数学：积分

本文主要参考&#xff1a; 【建议收藏】同济七版《高等数学》精讲视频 | 期末考试 | 考研零基础 | 高数小白_哔哩哔哩_bilibili 4.1.1.1 定积分引例_哔哩哔哩_bilibili 仅供本人学习使用。 积分是一个在多个领域都广泛使用的概念&#xff0c;特别是在数学和物理学中。 以下是积…
阅读更多...
深入了解c语言字符串 2

深入了解c语言字符串 2

深入了解c语言字符串 2 一 使用 scanf进行字符串的输入&#xff1a;1.1输入单词&#xff08;不包含空格&#xff09;&#xff1a;1.2 输入带空格的整行文本&#xff1a;1.3 处理输入缓冲区&#xff1a;1.4 注意安全性&#xff1a; 二 使用 printf 字符串的输出&#xff1a;三 输…
阅读更多...
春招秋招，在线测评到底难不难？

春招秋招，在线测评到底难不难？

现在很多企业在春招的时候&#xff0c;都会有一个在线测评的环节&#xff0c;目的当然就是希望更加了解清楚毕业生的综合能力以及其他方面的素质&#xff0c;好让HR可以根据岗位筛选出能力达标的人才。所以&#xff0c;现在不少即将面对春招的大学毕业生&#xff0c;比较关心的…
阅读更多...
uniapp中组件库Mask 遮罩层 的使用方法

uniapp中组件库Mask 遮罩层 的使用方法

目录 #平台差异说明 #基本使用 #嵌入内容 #遮罩样式 #API #Props #Events #Slot 创建一个遮罩层&#xff0c;用于强调特定的页面元素&#xff0c;并阻止用户对遮罩下层的内容进行操作&#xff0c;一般用于弹窗场景 #平台差异说明 AppH5微信小程序支付宝小程序百度小程…
阅读更多...
没有外网Nginx如何配置如何开启https

没有外网Nginx如何配置如何开启https

判断是否支持open-ssl 在服务器执行如下命令 openssl version没有则安装open-ssl&#xff0c;由于服务器没有外网&#xff0c;可以离线安装openssl-3.0.1.tar.gz&#xff0c;我是在有网的服务器直接下载的&#xff0c;然后再上传到这台无网的服务器上 wget https://www.open…
阅读更多...
C++内存管理与模板

C++内存管理与模板

C内存管理与模板 文章目录 C内存管理与模板前言&#xff1a;一.new和delete基本用法二.底层实现三.定位new四.模板4.1函数模板4.2调用选择4.3类模板4.4声明定义分离 前言&#xff1a; C的内存管理和C语言中动态内存分配是相似的&#xff0c;在这一篇我们会学到更符合面向对象的…
阅读更多...
如何快速写出高效的软件测试用例

如何快速写出高效的软件测试用例

&#x1f345; 视频学习&#xff1a;文末有免费的配套视频可观看 &#x1f345; 点击文末小卡片&#xff0c;免费获取软件测试全套资料&#xff0c;资料在手&#xff0c;涨薪更快 前言 编写测试用例的目的就是确保测试过程全面高效、有据可查。但要编写出高效的测试用例&#…
阅读更多...
鸿蒙harmony--TypeScript基础语法

鸿蒙harmony--TypeScript基础语法

把青春献给身后那座辉煌的都市&#xff0c;为了这个美梦我们付出着代价 目录 一&#xff0c;基础类型 二&#xff0c;数组 三&#xff0c;any 四&#xff0c;变量的类型注释 五&#xff0c;函数 5.1 参数类型注解 5.2 返回类型注解 5.3 匿名函数 六&#xff0c;对象类型 可选属…
阅读更多...
APP攻防-资产收集篇反证书检验XP框架反代理VPN数据转发反模拟器

APP攻防-资产收集篇反证书检验XP框架反代理VPN数据转发反模拟器

知识点 1、APP资产-抓包突破&反模拟器 2、APP资产-抓包突破&反证书检验 3、APP资产-抓包突破&反代理VPN 章节点&#xff1a; 1、APP资产-内在提取&外在抓包 2、APP逆向-反编译&删验证&重打包 3、APP安全-存储&服务&组件&注册等 专题点&…
阅读更多...
HDFS HA 之 HA 原理

HDFS HA 之 HA 原理

1 ZKFC解析 HA(High Availability)是HDFS支持的一个重要特性，可以有效解决Active Namenode遇到故障时，将可用的Standby节点变成新的Active状态的问题，使集群能够正常工作。目前支持冷切换和热切换两种方式。冷切换通过手动触发，缺点是不能够及时恢复集群。实际生产中以应用…
阅读更多...
2024年【危险化学品经营单位主要负责人】报名考试及危险化学品经营单位主要负责人免费试题

2024年【危险化学品经营单位主要负责人】报名考试及危险化学品经营单位主要负责人免费试题

题库来源&#xff1a;安全生产模拟考试一点通公众号小程序 危险化学品经营单位主要负责人报名考试参考答案及危险化学品经营单位主要负责人考试试题解析是安全生产模拟考试一点通题库老师及危险化学品经营单位主要负责人操作证已考过的学员汇总&#xff0c;相对有效帮助危险化…
阅读更多...
LeetCode：49. 字母异位词分组

LeetCode：49. 字母异位词分组

49. 字母异位词分组 1&#xff09;题目2&#xff09;代码3&#xff09;结果 1&#xff09;题目 给你一个字符串数组&#xff0c;请你将 字母异位词 组合在一起。可以按任意顺序返回结果列表。 字母异位词 是由重新排列源单词的所有字母得到的一个新单词。 示例 1: 输入: strs…
阅读更多...
python爬虫3

python爬虫3

1.异常处理&#xff0c;使代码更加健壮 静态cookie可视绕过登录的限制 快代理是一个代理平台 # https://movie.douban.com/j/chart/top_list?type5&interval_id100%3A90&action& # start0&limit20# https://movie.douban.com/j/chart/top_list?type5&int…
阅读更多...
【数据结构】 归并排序超详解

【数据结构】 归并排序超详解

1.基本思想 归并排序&#xff08;MERGE-SORT&#xff09;是建立在归并操作上的一种有效的排序算法,该算法是采用分治法&#xff08;Divide andConquer&#xff09;的一个非常典型的应用。 将已有序的子序列合并&#xff0c;得到完全有序的序列&#xff0c;即先使每个子序列有序…
阅读更多...
Vue3的Props

Vue3的Props

Vue 3中的props是用于接收父组件传递的数据的属性。在Vue 3中&#xff0c;props的声明发生了一些改变&#xff1a; 使用props选项来声明props。之前的版本中使用props属性来声明&#xff0c;但在Vue 3中改为使用props选项。通过TypeScript或Flow来静态类型检查props。Vue 3允许…
阅读更多...
Spring实现事务(一)

Spring实现事务(一)

Spring事务 .什么是事务事务的操作Spring中事务的实现准备工作创建表创建项目,引入Spring Web, Mybatis, mysql等依赖配置文件实体类 编程式事务(手动写代码操作事务)声明式事务(利用注解自动开启和提交事务) . 什么是事务 事务是⼀组操作的集合, 是⼀个不可分割的操作 在我们…
阅读更多...
国产校准件

国产校准件

国产校准件 Ceyear系列校准件是矢量网络分析仪的测试附件&#xff0c;可大幅提高矢量网络分析仪的测试精度。规格品种丰富&#xff0c;涵盖多种同轴、波导校准件&#xff0c;校准精度高&#xff0c;重复性好 功能特点 校准件 校准模块可实现更精准的测量&#xff0c;满足您的测…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023