大数据(9g)FlinkCEP-创新互联
文章目录
概述
网页标题:大数据(9g)FlinkCEP-创新互联
本文地址:http://pwwzsj.com/article/dgicho.html
- 概述
- 示例代码
- 环境和依赖
- Java代码
- 上面代码可改成下面
概述
- CEP
Complex Event Processing:复合事件处理
通过分析事件间的关系,从事件流中查询出符合要求的事件序列 - 例如【切菜=>洗菜=>炒菜】3个事件按时间序串联,是正常的事件流
当发现【切菜=>炒菜】忽略洗菜的事件流,可认为是异常事件
WIN10+JDK1.8+IDEA2021+Maven3.6.3
CEP额外依赖为flink-cep
8 8 1.14.6 2.12 1.18.24 org.apache.flink flink-java ${flink.version} org.apache.flink flink-streaming-java_${scala.binary.version} ${flink.version} org.apache.flink flink-clients_${scala.binary.version} ${flink.version} org.apache.flink flink-runtime-web_${scala.binary.version} ${flink.version} org.apache.flink flink-cep_${scala.binary.version} ${flink.version}
Java代码监测 严格近邻的连续三次a的事件流
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class CepPractice {public static void main(String[] args) throws Exception {//创建环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
//添加数据源,确定水位线策略
SingleOutputStreamOperatord = env.fromElements("c", "a", "a", "a", "a", "b", "a", "a")
.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps()
.withTimestampAssigner((element, recordTimestamp) ->1L));
//定义模式
Patternp = Pattern
.begin("first")
.where(new SimpleCondition() {@Override
public boolean filter(String value) {return value.equals("a");
}
})
.next("second")
.where(new SimpleCondition() {@Override
public boolean filter(String value) {return value.equals("a");
}
})
.next("third")
.where(new SimpleCondition() {@Override
public boolean filter(String value) {return value.equals("a");
}
});
//在流上匹配模型
PatternStreampatternStream = CEP.pattern(d, p);
//使用select方法将匹配到的事件流取出
patternStream.select((PatternSelectFunction) map ->{//Map的key是事件名称(上面的first、second和third)
//Map的key对应的value是列表,储存匹配到的事件
String first = map.get("first").toString();
String second = map.get("second").toString();
String third = map.get("third").toString();
return first + "->" + second + "->" + third;
}).print();
//执行
env.execute();
}
}
打印结果
[a]->[a]->[a]
[a]->[a]->[a]
上面代码可改成下面留意.times(3).consecutive()
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.List;
public class CepPractice2 {public static void main(String[] args) throws Exception {//创建环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
//添加数据源,确定水位线策略
SingleOutputStreamOperator>d = env.fromElements(
Tuple2.of("a", 1000L), Tuple2.of("a", 2000L), Tuple2.of("a", 3000L),
Tuple2.of("a", 4000L), Tuple2.of("b", 5000L), Tuple2.of("a", 6000L))
.assignTimestampsAndWatermarks(WatermarkStrategy.>forMonotonousTimestamps()
.withTimestampAssigner((element, recordTimestamp) ->element.f1));
//定义模式
Pattern, Tuple2>p = Pattern
.>begin("=a")
.where(new SimpleCondition>() {@Override
public boolean filter(Tuple2value) {return value.f0.equals("a");
}
})
.times(3)
.consecutive(); //严格连续
//在流上匹配模型
PatternStream>patternStream = CEP.pattern(d, p);
//使用select方法将匹配到的事件流取出
patternStream.select((PatternSelectFunction, String>) map ->{//Map的key是事件名称(上面的first、second和third)
//Map的key对应的value是列表,储存匹配到的事件
List>ls = map.get("=a");
String first = ls.get(0).f0;
String second = ls.get(1).f0;
String third = ls.get(2).f0;
return String.join("=>", first, second, third);
}).print();
//执行
env.execute();
}
}
你是否还在寻找稳定的海外服务器提供商?创新互联www.cdcxhl.cn海外机房具备T级流量清洗系统配攻击溯源,准确流量调度确保服务器高可用性,企业级服务器适合批量采购,新人活动首月15元起,快前往官网查看详情吧
网页标题:大数据(9g)FlinkCEP-创新互联
本文地址:http://pwwzsj.com/article/dgicho.html