今日指数项目之FlinkCEP入门案例
Posted Maynor学长
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了今日指数项目之FlinkCEP入门案例相关的知识，希望对你有一定的参考价值。
CEP案例
1.入门案例
需求：
有一个业务系统，用户要使用该业务系统必须要先登陆
过滤出来在2秒内连续登陆失败的用户
在test源码目录下创建测试类：cn.itcast.LoginFailDemo
开发步骤：
1.获取流处理执行环境
2.设置并行度,设置事件时间
加载数据源,提取事件时间
4.定义匹配模式，设置时间长度
5.匹配模式（分组）
6.数据处理
7.打印
8.触发执行
数据源：
Arrays.asList(
new LoginUser (1, "192.168.0.1", "fail", 1558430842000L), //2019-05-21 17:27:22
new LoginUser (1, "192.168.0.2", "fail", 1558430843000L), //2019-05-21 17:27:23
new LoginUser (1, "192.168.0.3", "fail", 1558430844000L), //2019-05-21 17:27:24
new LoginUser (2, "192.168.10.10", "success", 1558430845000L) //2019-05-21 17:27:25
)
参考代码
/**
* 使用CEP实现三秒内登录失败两次的用户
*/
public class LoginFailDemo
public static void main(String[] args) throws Exception
//1：初始化流式运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//2：设置并行度为1
env.setParallelism(1);
//3：指定数据按照事件时间进行处理
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//4：构建数据源
DataStream<LoginUser > LoginUserStream = env.fromCollection(Arrays.asList(
new LoginUser (1, "192.168.0.1", "fail", 1558430842000L),//2019-05-21 17:27:22
new LoginUser (1, "192.168.0.2", "fail", 1558430843000L),//2019-05-21 17:27:23
new LoginUser (1, "192.168.0.3", "fail", 1558430844000L),//2019-05-21 17:27:24
new LoginUser (2, "192.168.10.10", "success", 1558430845000L)//2019-05-21 17:27:25
)).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LoginUser>(Time.seconds(0))
@Override
public long extractTimestamp(LoginUser element)
return element.getEventTime();
);
//5.1：定义规则模型
Pattern<LoginUser, LoginUser > LoginUserPattern = Pattern.<LoginUser >begin("begin")
.where(new IterativeCondition<LoginUser>()
@Override
public boolean filter(LoginUser loginUser, Context<LoginUser > context) throws Exception
return loginUser.getEventType().equals("fail");
)//匹配第一个事件，匹配的是登录失败
.next("next") //匹配到第一个事件以后，紧跟着一个事件数据，next表示两个事件必须严格的临近
.where(new IterativeCondition<LoginUser >()
@Override
public boolean filter(LoginUser loginUser, Context<LoginUser> context) throws Exception
return loginUser.getEventType().equals("fail");
)//匹配第二个事件，匹配的是登录失败
.within(Time.seconds(3));//定义结束状态，结束状态可以是时间触发也可以是满足了某个事件触发
//5.2：将规则模型应用到数据流中
PatternStream<LoginUser > patternDataStream = CEP.pattern(LoginUserStream.keyBy(LoginUser ::getUserId), LoginUserPattern);
//5.3：获取到符合规则模型的数据
/**
* IN：传入的数据类型
* OUT：返回值的数据类型
* (Long, String, String, Long):(用户id, 登录ip，登录状态，登录时间)
*/
SingleOutputStreamOperator<Tuple4<Integer, String, String, Long>> loginFailDataStream = patternDataStream.select(new PatternSelectFunction<LoginUser, Tuple4<Integer, String, String, Long>>()
@Override
public Tuple4<Integer, String, String, Long> select(Map<String, List<LoginUser>> pattern) throws Exception
//根据刚才的分析，符合规则的数据会存储到状态集合中，也就是state中，所以查找匹配的时候需要在state中获取数据
LoginUser loginUser = pattern.getOrDefault("next", null).iterator().next();
//返回匹配到的数据
return Tuple4.of(loginUser.getUserId(), loginUser.getIp(), loginUser.getEventType(), loginUser.getEventTime());
);
//打印出来符合条件的数据
loginFailDataStream.print("连续两次登录失败的用户>>>");
//执行任务
env.execute();
登陆对象：
public int userId; //用户id
public String ip;//用户Ip
public String eventType; //状态
public Long eventTime;//事件时间
/**
* 构建登录对象
*/
public static class LoginUser implements Serializable
public int userId; //用户id
public String ip;//用户Ip
public String eventType; //状态
public Long eventTime;//事件时间
public int getUserId()
return userId;
public void setUserId(int userId)
this.userId = userId;
public String getIp()
return ip;
public void setIp(String ip)
this.ip = ip;
public String getEventType()
return eventType;
public void setEventType(String eventType)
this.eventType = eventType;
public Long getEventTime()
return eventTime;
public void setEventTime(Long eventTime)
this.eventTime = eventTime;
public LoginEvent(int userId, String ip, String eventType, Long eventTime)
this.userId = userId;
this.ip = ip;
this.eventType = eventType;
this.eventTime = eventTime;
@Override
public String toString()
return "LoginEvent" +
"userId=" + userId +
", ip='" + ip + '\\'' +
", eventType='" + eventType + '\\'' +
", eventTime=" + eventTime +
'';
2.监控市场价格
需求：
物价局和工商局会监督市场上各种商品得销售价格，随着市场行情和商品供需得变化，商品价格会有一定程度得浮动，如果商品价格在指定得价格区间波动，政府部门是不会干预的额，如果商品价格在一定的时间范围内波动幅度超出了指定的区间范围，并且上行幅度过大，物价局会上报敏感数据信息，并规范市场价格。
在此，我们假定如果商品售价在1分钟之内有连续两次超过预定商品价格阀值就发送告警信息。
测试数据
"goodsId":100001,"goodsPrice":6,"goodsName":"apple","alias":"苹果","orderTime":1558430843000
"goodsId":100007,"goodsPrice":0.5,"goodsName":"mask","alias":"口罩","orderTime":1558430844000
"goodsId":100002,"goodsPrice":2,"goodsName":"rice","alias":"大米","orderTime":1558430845000
"goodsId":100003,"goodsPrice":2,"goodsName":"flour","alias":"面粉","orderTime":1558430846000
"goodsId":100004,"goodsPrice":12,"goodsName":"rice","alias":"大米","orderTime":1558430847000
"goodsId":100005,"goodsPrice":20,"goodsName":"apple","alias":"苹果","orderTime":1558430848000
"goodsId":100006,"goodsPrice":3,"goodsName":"banana","alias":"香蕉","orderTime":1558430849000
"goodsId":100007,"goodsPrice":10,"goodsName":"mask","alias":"口罩","orderTime":1558430850000
"goodsId":100001,"goodsPrice":16,"goodsName":"apple","alias":"苹果","orderTime":1558430852000
"goodsId":100007,"goodsPrice":15,"goodsName":"mask","alias":"口罩","orderTime":1558430853000
"goodsId":100002,"goodsPrice":12,"goodsName":"rice","alias":"大米","orderTime":1558430854000
"goodsId":100003,"goodsPrice":12,"goodsName":"flour","alias":"面粉","orderTime":1558430855000
"goodsId":100004,"goodsPrice":12,"goodsName":"rice","alias":"大米","orderTime":1558430856000
"goodsId":100005,"goodsPrice":20,"goodsName":"apple","alias":"苹果","orderTime":1558430857000
"goodsId":100006,"goodsPrice":13,"goodsName":"banana","alias":"香蕉","orderTime":1558430858000
"goodsId":100007,"goodsPrice":10,"goodsName":"mask","alias":"口罩","orderTime":1558430859000
创建kafka topic
./kafka-topics.sh --create --topic cep --zookeeper node01:2181 --partitions 1 --replication-factor 1
生产数据
./kafka-console-producer.sh --broker-list node01:9092 --topic cep
redis保存限制价格
jedisCluster.hset(“product”,“apple”,“10”);
jedisCluster.hset(“product”,“rice”,“6”);
jedisCluster.hset(“product”,“flour”,“6”);
jedisCluster.hset(“product”,“banana”,“8”);
jedisCluster.hset(“product”,“mask”,“5”);
开发步骤
在test源码目录下创建测试类：cn.itcast.CepMarkets
1.获取流处理执行环境
2.设置事件时间、并行度
整合kafka
4.数据转换
5.process获取bean,设置status，并设置事件时间
6.定义匹配模式，设置时间长度
7.匹配模式（分组）
8.查询告警数据
2.1.代码开发
public class CepMarkets
public static void main(String[] args) throws Exception
//1.获取流处理执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
//2.设置事件时间
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//3.整合kafka
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "node01:9092"); //broker地址
properties.setProperty("group.id", "cep"); //消费组
properties.setProperty("enable.auto.commit", "true");
properties.setProperty("auto.commit.interval.ms", "5000");
FlinkKafkaConsumer011<String> kafkaConsumer = new FlinkKafkaConsumer011<>("cep", new SimpleStringSchema(), properties);
kafkaConsumer.setStartFromEarliest();
DataStreamSource<String> source = env.addSource(kafkaConsumer);
//4.数据转换
SingleOutputStreamOperator<Product> mapData = source.map(new MapFunction<String, Product>()
@Override
public Product map(String value) throws Exception
JSONObject json = JSON.parseObject(value);
Product product = new Product(
json.getLong("goodsId"),
json.getDouble("goodsPrice"),
json.getString("goodsName"),
json.getString("alias"),
json.getLong("orderTime"),
false
);
return product;
);
//5.保留告警数据(设置时间)
SingleOutputStreamOperator<Product> waterData = mapData.keyBy(Product::getGoodsId)
.process(new KeyedProcessFunction<Long, Product, Product>()
Map<String, String> map = null;
@Override
public void open(Configuration parameters) throws Exception
JedisCluster jedisCluster = RedisUtil.getJedisCluster();
map = jedisCluster.hgetAll("product");
@Override
public void processElement(Product value, Context ctx, Collector<Product> out) throws Exception
long priceAlert = Long.parseLong(map.get(value.getGoodsName()));
if (value.getGoodsPrice() > priceAlert)
value.setStatus(true);
out.collect(value);
)
.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Product>(Time.seconds(0))
@Override
public long extractTimestamp(Product element)
return element.getOrderTime();
)
;
//6.定义匹配模式，设置时间长度
Pattern<Product, Product> pattern = Pattern.<Product>begin("begin")
.where(new SimpleCondition<Product>()
@Override
public boolean filter(Product value) throws Exception
return value.getStatus() == true;
)
.next("next")
.where(new SimpleCondition<Product>()
@Override
public boolean filter(Product value) throws Exception
return value.getStatus() == true;
)
.within(Time.seconds(60));
//7.匹配模式（分组）
PatternStream<Product> cep = CEP.pattern(waterData.keyBy(Product::getGoodsId), pattern);
//8.查询告警数据
cep.select(new PatternSelectFunction<Product, Object>()
@Override
public Object select(Map<String, List<Product>> pattern) throws Exception
List<Product> result = pattern.get("next");
return result;
).print("告警数据：");
env.execute();
2.2.Bean对象
属性：goodsId、goodsPrice、goodsName、alias、orderTime、status
public class Product
private Long goodsId;
private Double goodsPrice;
private String goodsName;
private String alias;
private Long orderTime;
private Boolean status;
public Product(Long goodsId, Double goodsPrice, String goodsName, String alias, Long orderTime, Boolean status)
this.goodsId = goodsId;
this.goodsPrice = goodsPrice;
this.goodsName = goodsName;
this.alias = alias;
this.orderTime = orderTime;
this.status = status;
@Override
public String toString()
return "Product" +
"goodsId=" + goodsId +
", goodsPrice=" + goodsPrice +
", goodsName='" + goodsName + '\\'' +
", alias='" + alias + '\\'' +
", orderTime=" + orderTime +
", status=" + status +
'';
public Long getGoodsId()
return goodsId;
public void setGoodsId(Long goodsId)
this.goodsId = goodsId;
public Double getGoodsPrice()
return goodsPrice;
public void setGoodsPrice(Double goodsPrice)
this.goodsPrice = goodsPrice;
public String getGoodsName()
return goodsName;
public void setGoodsName(String goodsName)
this.goodsName = goodsName;
public String getAlias()
return alias;
public void setAlias(String alias)
this.alias = alias;
public Long getOrderTime()
return orderTime;
public void setOrderTime(Long orderTime)
this.orderTime = orderTime;
public Boolean getStatus()
return status;
public void setStatus(Boolean status)
this.status = status;
以上是关于今日指数项目之FlinkCEP入门案例的主要内容，如果未能解决你的问题，请参考以下文章