CEP 即Complex Event Processing - 复杂事件，Flink CEP 是在 Flink 中实现的复杂时间处理(CEP)库。处理事件的规则，被叫做“模式”(Pattern)，Flink CEP 提供了 Pattern API，用于对输入流数据进行复杂事件规则定义，用来提取符合规则的事件序列。
Pattern API 大致分为三种：个体模式，组合模式，模式组。(详情请自行查看资料)
Flink CEP 应用场景：
CEP 在互联网各个行业都有应用，例如金融、物流、电商、智能交通、物联网行业等行业：
实时监控：我们需要在大量的订单交易中发现那些虚假交易，在网站的访问日志中寻找那些使用脚本或者工具“爆破”登录的用户，或者在快递运输中发现那些滞留很久没有签收的包裹等。
风险控制：比如金融行业可以用来进行风险控制和欺诈识别，从交易信息中寻找那些可能存在危险交易和非法交易。
营销广告：跟踪用户的实时行为，指定对应的推广策略进行推送，提高广告的转化率。
Flink CEP 开发流程：
- DataSource 中的数据转换为 DataStream；
- 定义 Pattern，并将 DataStream 和 Pattern 组合转换为 PatternStream；
- PatternStream 经过 select、process 等算子转换为 DataStraem；
- 再次转换的 DataStream 经过处理后，sink 到目标库。
接下来我们讲对 超时未支付、连续登录、交易活跃用户 这三个场景进行实操。
一、超时未支付
需求：找出那些下单后 10 分钟内没有支付的订单。
- 创建 TimeOutPayCEPMain.java 类并编写整体代码：
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<PayEvent> source = env.fromElements(
new PayEvent(1L, "create", 1597905234000L),
new PayEvent(1L, "pay", 1597905235000L),
new PayEvent(2L, "create", 1597905236000L),
new PayEvent(2L, "pay", 1597905237000L),
new PayEvent(3L, "create", 1597905239000L)
).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<PayEvent>(Time.milliseconds(500L)) {
@Override
public long extractTimestamp(PayEvent payEvent) {
return payEvent.getTimestamp();
}
}).keyBy(new KeySelector<PayEvent, Object>() {
@Override
public Object getKey(PayEvent value) throws Exception {
return value.getUserId();
}
});
// 逻辑处理代码
env.execute("execute cep");
}
- 逻辑处理代码如下：
OutputTag<PayEvent> orderTimeoutOutput = new OutputTag<PayEvent>("orderTimeout") {};
Pattern<PayEvent, PayEvent> pattern = Pattern.<PayEvent>
begin("begin")
.where(new IterativeCondition<PayEvent>() {
@Override
public boolean filter(PayEvent payEvent, Context context) throws Exception {
return payEvent.getActive().equals("create");
}
})
.followedBy("pay")
.where(new IterativeCondition<PayEvent>() {
@Override
public boolean filter(PayEvent payEvent, Context context) throws Exception {
return payEvent.getActive().equals("pay");
}
})
.within(Time.seconds(600));
PatternStream<PayEvent> patternStream = CEP.pattern(source, pattern);
SingleOutputStreamOperator<PayEvent> result = patternStream.select(orderTimeoutOutput, new PatternTimeoutFunction<PayEvent, PayEvent>() {
@Override
public PayEvent timeout(Map<String, List<PayEvent>> map, long l) throws Exception {
return map.get("begin").get(0);
}
}, new PatternSelectFunction<PayEvent, PayEvent>() {
@Override
public PayEvent select(Map<String, List<PayEvent>> map) throws Exception {
return map.get("pay").get(0);
}
});
//result.print();
DataStream<PayEvent> sideOutput = result.getSideOutput(orderTimeoutOutput);
sideOutput.print();
-
运行结果：image.png
二、连续登录失败
需求：找出那些 5 秒钟内连续登录失败的账号，然后禁止用户再次尝试登录需要等待 1 分钟。
- 创建 LoginCEPMain.java 类并实现前提代码：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<LogInEvent> source = env.fromElements(
new LogInEvent(1L, "fail", 1597905234000L),
new LogInEvent(1L, "success", 1597905235000L),
new LogInEvent(2L, "fail", 1597905236000L),
new LogInEvent(2L, "fail", 1597905237000L),
new LogInEvent(2L, "fail", 1597905238000L),
new LogInEvent(3L, "fail", 1597905239000L),
new LogInEvent(3L, "success", 1597905240000L)
).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LogInEvent>(Time.milliseconds(500L)) {
@Override
public long extractTimestamp(LogInEvent logInEvent) {
return logInEvent.getTimestamp();
}
}).keyBy(new KeySelector<LogInEvent, Object>() {
@Override
public Object getKey(LogInEvent value) throws Exception {
return value.getuId();
}
});
// 关键逻辑代码
env.execute("execute cep");
- 关键逻辑代码 如下：
Pattern pattern = Pattern.<LogInEvent>begin("start").where(new IterativeCondition<LogInEvent>() {
@Override
public boolean filter(LogInEvent logInEvent, Context<LogInEvent> context) throws Exception {
return logInEvent.getAction().equals("fail");
}
}).next("next").where(new IterativeCondition<LogInEvent>() {
@Override
public boolean filter(LogInEvent logInEvent, Context<LogInEvent> context) throws Exception {
return logInEvent.getAction().equals("fail");
}
}).within(Time.seconds(5));
PatternStream<LogInEvent> patternStream = CEP.pattern(source, pattern);
SingleOutputStreamOperator<AlertEvent> process = patternStream.process(new PatternProcessFunction<LogInEvent, AlertEvent>() {
@Override
public void processMatch(Map<String, List<LogInEvent>> match, Context ctx, Collector<AlertEvent> out) throws Exception {
List<LogInEvent> start = match.get("start");
List<LogInEvent> next = match.get("next");
System.out.println("start:" + start + ",next:" + next);
out.collect(new AlertEvent(String.valueOf(start.get(0).getuId()), "Login fail ~ "));
}
});
process.print();
-
运行结果：image.png
三、交易活跃用户
需求：找出那些 24 小时内至少 5 次有效交易的账户。
- 创建 ActiveCEPMain.java 类并实现前提代码：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<TransactionEvent> source = env.fromElements(
new TransactionEvent("100XX", 0.0D, 1597905234000L),
new TransactionEvent("100XX", 100.0D, 1597905235000L),
new TransactionEvent("100XX", 200.0D, 1597905236000L),
new TransactionEvent("100XX", 300.0D, 1597905237000L),
new TransactionEvent("100XX", 400.0D, 1597905238000L),
new TransactionEvent("100XX", 500.0D, 1597905239000L),
new TransactionEvent("101XX", 0.0D, 1597905240000L),
new TransactionEvent("101XX", 100.0D, 1597905241000L)
).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<TransactionEvent>(Time.milliseconds(500L)) {
@Override
public long extractTimestamp(TransactionEvent transactionEvent) {
return transactionEvent.getTimestamp();
}
}).keyBy(new KeySelector<TransactionEvent, Object>() {
@Override
public Object getKey(TransactionEvent value) throws Exception {
return value.getAccountId();
}
});
// 关键逻辑处理代码
env.execute("execute cep");
- 关键逻辑处理代码 ：
Pattern pattern = Pattern.<TransactionEvent>begin("start").where(
new SimpleCondition<TransactionEvent>() {
@Override
public boolean filter(TransactionEvent transactionEvent) {
return transactionEvent.getAmount() > 0;
}
}
).timesOrMore(5)
.within(Time.hours(24));
PatternStream<TransactionEvent> patternStream = CEP.pattern(source, pattern);
SingleOutputStreamOperator<AlertEvent> process = patternStream.process(new PatternProcessFunction<TransactionEvent, AlertEvent>() {
@Override
public void processMatch(Map<String, List<TransactionEvent>> map, Context context, Collector<AlertEvent> collector) throws Exception {
List<TransactionEvent> start = map.get("start");
List<TransactionEvent> next = map.get("next");
System.out.println("start:" + start + ",next:" + next);
collector.collect(new AlertEvent(start.get(0).getAccountId(), "lian xu 有效交易！"));
}
});
process.print();
-
运行结果：image.png
到此，我们的三种场景的案例已经操作完毕，从中可以看出 CEP 技术开发流程以及模式的定义之后其实就不算难，这些玩意主要还是靠理解，多练习才能熟练应用的；好了，本次总结就到这里，希望能帮助到你哦同学。
