基于Flink+kafka实时告警

这篇具有很好参考价值的文章主要介绍了基于Flink+kafka实时告警。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

引出问题

项目使用告警系统的逻辑是将实时数据保存到本地数据库再使用定时任务做判断,然后产生告警数据。这种方式存在告警的延时实在是太高了。数据从产生到保存,从保存到判断都会存在时间差,按照保存数据定时5分钟一次,定时任务5分钟一次。最高会产生10分钟的误差,这种告警就没什么意义了。

demo设计

为了简单的还原业务场景,做了简单的demo假设

实现一个对于学生成绩评价的实时处理程序
数学成绩,基准范围是90-140,超出告警
物理成绩,基准范围是60-95,超出告警

环境搭建

使用windows环境演示

准备工作

1、安装jdk

2、安装zookeeper

解压压缩包

zoo_sample.cfg将它重命名为zoo.cfg

修改配置 dataDir=D://tools//apache-zookeeper-3.5.10-bin//data

配置环境变量

3、安装kafka

解压压缩包

修改config/server.properties

log.dirs=D://tools//kafka_2.11-2.1.0//log

flink程序代码

pom

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>1.13.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.12</artifactId>
    <version>1.13.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients_2.12</artifactId>
    <version>1.13.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.12</artifactId>
    <version>1.13.0</version>
</dependency>

<!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>1.18.12</version>
    <scope>provided</scope>
</dependency>

<!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.62</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.11</artifactId>
    <version>1.10.0</version>
</dependency>

主程序

public class StreamAlertDemo {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(3);
		Properties properties = new Properties();
		properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
		FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties);
		DataStreamSource<String> inputDataStream = env.addSource(kafkaConsumer);

		DataStream<String> resultStream = inputDataStream.flatMap(new AlertFlatMapper());
		resultStream.print().setParallelism(4);

		resultStream.addSink(new FlinkKafkaProducer<>("demo",new SimpleStringSchema(),properties));
		env.execute();
	}

}
主程序,配置告警规则后期可以使用推送或者拉去方式获取数据
public class RuleMap {

	private RuleMap(){}

	public final static Map<String,List<AlertRule>> initialRuleMap;

	private static List<AlertRule> ruleList = new ArrayList<>();

	private static List<String> ruleStringList = new ArrayList<>(Arrays.asList(
			"{\"target\":\"MathVal\",\"type\":\"0\",\"criticalVal\":90,\"descInfo\":\"You Math score is too low\"}",
			"{\"target\":\"MathVal\",\"type\":\"2\",\"criticalVal\":140,\"descInfo\":\"You Math score is too high\"}",
			"{\"target\":\"PhysicsVal\",\"type\":\"0\",\"criticalVal\":60,\"descInfo\":\"You Physics score is too low\"}",
			"{\"target\":\"PhysicsVal\",\"type\":\"2\",\"criticalVal\":95,\"descInfo\":\"You Physics score is too high\"}"));

	static {
		for (String i : ruleStringList) {
			ruleList.add(JSON.parseObject(i, AlertRule.class));
		}
		initialRuleMap = ruleList.stream().collect(Collectors.groupingBy(AlertRule::getTarget));
	}


}

AlertFlatMapper,处理告警逻辑

public class AlertFlatMapper implements FlatMapFunction<String, String> {

	@Override
	public void flatMap(String inVal, Collector<String> out) throws Exception {
		Achievement user = JSON.parseObject(inVal, Achievement.class);
		Map<String, List<AlertRule>> initialRuleMap = RuleMap.initialRuleMap;
		List<AlertInfo> resList = new ArrayList<>();
		List<AlertRule> mathRule = initialRuleMap.get("MathVal");
		for (AlertRule rule : mathRule) {
			if (checkVal(user.getMathVal(), rule.getCriticalVal(), rule.getType())) {
				resList.add(new AlertInfo(user.getName(), rule.getDescInfo()));
			}
		}
		List<AlertRule> physicsRule = initialRuleMap.get("PhysicsVal");
		for (AlertRule rule : physicsRule) {
			if (checkVal(user.getPhysicsVal(), rule.getCriticalVal(), rule.getType())) {
				resList.add(new AlertInfo(user.getName(), rule.getDescInfo()));
			}
		}
		String result = JSON.toJSONString(resList);
		out.collect(result);
	}

	private static boolean checkVal(Integer actVal, Integer targetVal, Integer type) {
		switch (type) {
			case 0:
				return actVal < targetVal;
			case 1:
				return actVal.equals(targetVal);
			case 2:
				return actVal > targetVal;
			default:
				return false;
		}
	}
}

三个实体类

@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
public class Achievement implements Serializable {

    private static final long serialVersionUID = -1L;

    private String name;

    private Integer mathVal;

    private Integer physicsVal;

}

@Data
@AllArgsConstructor
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
public class AlertInfo implements Serializable {
    
    private static final long serialVersionUID = -1L;

    private String name;

    private String descInfo;

}

@Data
@AllArgsConstructor
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
public class AlertRule implements Serializable {

	private static final long serialVersionUID = -1L;

	private String target;

	//0小于 1等于 2大于
	private Integer type;

	private Integer criticalVal;

	private String descInfo;
}

 项目演示

创建kafka生产者 test
.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test

创建kafka消费者 demo
.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic demo --from-beginning

启动flink应用

给topic test发送消息

{"name":"liu","MathVal":45,"PhysicsVal":76}

 消费topic demo

flink 告警规则,项目设计,flink,大数据

告警系统架构

flink 告警规则,项目设计,flink,大数据文章来源地址https://www.toymoban.com/news/detail-595584.html

到了这里,关于基于Flink+kafka实时告警的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 基于华为MRS3.2.0实时Flink消费Kafka落盘至HDFS的Hive外部表的调度方案

    该需求为实时接收对手Topic,并进行消费落盘至Hive。 在具体的实施中,基于华为MRS 3.2.0安全模式带kerberos认证的Kafka2.4、Flink1.15、Hadoop3.3.1、Hive3.1,调度平台为开源dolphinscheduler。 本需求的完成全部参考华为官方MRS3.2.0开发文档,相关章节是普通版的安全模式。 华为官方文档:

    2024年01月21日
    浏览(39)
  • 【FLINK】Kafka数据源通过Flink-cdc进行实时数据同步

    CDC是Change Data Capture的缩写,中文意思是 变更数据获取 ,flink-cdc的作用是,通过flink捕获数据源的事务变动操作记录,包括数据的增删改操作等,根据这些记录可作用于对目标端进行实时数据同步。 下图是flink-cdc最新支持的数据源类型: kafka的数据源要通过flink-cdc进行实时数

    2024年02月12日
    浏览(58)
  • Flink CDC+Kafka 加速业务实时化

    摘要: 本文整理自阿里巴巴开发工程师,Apache Flink Committer 任庆盛,在 9 月 24 日 Apache Flink Meetup 的分享。主要内容包括: Flink CDC 技术对比与分析 Flink + Kafka 实时数据集成方案 Demo:Flink+Kafka 实现 CDC 数据的实时集成和实时分析 1.1. 变更数据捕获(CDC)技术 广义概念上,能够

    2024年02月15日
    浏览(49)
  • flink日志实时采集写入Kafka/ElasticSearch

    由于公司想要基于flink的日志做实时预警功能,故需要实时接入,并刷入es进行分析。 日志接入必须异步,不能影响服务性能 kafka集群宕机,依旧能够提交flink任务且运行任务 kafka集群挂起恢复,可以依旧续写实时运行日志 在类上加上@Plugin注解,标记为自定义appender 在类加上

    2024年02月08日
    浏览(54)
  • flink规则引擎设计思路

    在日常工作中我们经常收到一些诸如此类需求:“用户给点击了开屏广告,给用户下发私信”、“用户进入了推荐线,但在60秒内没有任何点击操作,弹框引导用户选择感兴趣的内容”、“用户点赞了某位作者的两篇以上的内容,但并没有关注过此作者,则弹框引导用户关注

    2024年02月05日
    浏览(40)
  • 流批一体计算引擎-4-[Flink]消费kafka实时数据

    Python3.6.9 Flink 1.15.2消费Kafaka Topic PyFlink基础应用之kafka 通过PyFlink作业处理Kafka数据 PyFlink需要特定的Python版本,Python 3.6, 3.7, 3.8 or 3.9。 1.3.1 python3和pip3的配置 一、系统中安装了多个版本的python3 。 二、环境变量path作用顺序 三、安装Pyflink 1.3.2 配置Flink Kafka连接 (1)在https://mvnr

    2024年02月06日
    浏览(43)
  • 实时Flink的数据库与Kafka集成优化案例

    在现代数据处理系统中,实时数据处理和分析是至关重要的。Apache Flink是一个流处理框架,可以用于实时数据处理和分析。在许多场景下,Flink需要与数据库和Kafka等消息系统进行集成,以实现更高效的数据处理。本文将讨论Flink与数据库和Kafka集成的优化案例,并提供实际示

    2024年02月20日
    浏览(43)
  • 掌握实时数据流:使用Apache Flink消费Kafka数据

            导读:使用Flink实时消费Kafka数据的案例是探索实时数据处理领域的绝佳方式。不仅非常实用,而且对于理解现代数据架构和流处理技术具有重要意义。         Apache Flink  是一个在 有界 数据流和 无界 数据流上进行有状态计算分布式处理引擎和框架。Flink 设计旨

    2024年02月03日
    浏览(81)
  • 实战Flink Java api消费kafka实时数据落盘HDFS

    在Java api中,使用flink本地模式,消费kafka主题,并直接将数据存入hdfs中。 flink版本1.13 kafka版本0.8 hadoop版本3.1.4 为了完成 Flink 从 Kafka 消费数据并实时写入 HDFS 的需求,通常需要启动以下组件: 确保 Zookeeper 在运行,因为 Flink 的 Kafka Consumer 需要依赖 Zookeeper。 确保 Kafka Serve

    2024年01月24日
    浏览(51)
  • 构建高效实时数据流水线:Flink、Kafka 和 CnosDB 的完美组合

    当今的数据技术生态系统中,实时数据处理已经成为许多企业不可或缺的一部分。为了满足这种需求,Apache Flink、Apache Kafka和CnosDB等开源工具的结合应运而生,使得实时数据流的收集、处理和存储变得更加高效和可靠。本篇文章将介绍如何使用 Flink、Kafka 和 CnosDB 来构建一个

    2024年02月10日
    浏览(43)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包