Flink cep需求代码实践02

╰半夏微凉° 2022-11-15 11:24 160阅读 0赞

# 业务需求01：  #

疑惑点：关于水印的环节，最后一条数据不进入cep触发计算。

>     初始operation_type=1且接下来的数据也是operation_type=1，且次数大于等于3，并且数据与数据之前的时间间隔要小于11000毫秒，且所有匹配数据的时间间隔必须在5分钟之内。

数据如下：

<table> 
 <tbody> 
  <tr> 
   <td>{"operation_type":"1","order_number":"1","device_code":"6f253e518373bcbe","create_date":"1616653174000"}</td> 
  </tr> 
  <tr> 
   <td>{"operation_type":"1","order_number":"2","device_code":"6f253e518373bcbe","create_date":"1616653274000"}</td> 
  </tr> 
  <tr> 
   <td>{"operation_type":"1","order_number":"3","device_code":"6f253e518373bcbe","create_date":"1616653374000"}</td> 
  </tr> 
  <tr> 
   <td>{"operation_type":"1","order_number":"4","device_code":"6f253e518373bcbe","create_date":"1616653474000"}</td> 
  </tr> 
  <tr> 
   <td>{"operation_type":"2","order_number":"5","device_code":"6f253e518373bcbe","create_date":"1616653574000"}</td> 
  </tr> 
  <tr> 
   <td>{"operation_type":"1","order_number":"6","device_code":"6f253e518373bcbe","create_date":"1616653674000"}</td> 
  </tr> 
 </tbody> 
</table>

代码如下：

>     package com.test.yarn;
>     
>     /**
>      * @program: flink-cep_2.11
>      * @description:
>     * @author: Mr.Wang
>      * @create: 2020-12-09 09:58
>      **/
>     
>     
>     import com.alibaba.fastjson.JSONObject;
>     import com.sjb.constant.Constants;
>     import com.sjb.schema.CustomerDeserializationSchema;
>     import com.sjb.test.HBaseWriter;
>     import org.apache.flink.api.common.functions.FilterFunction;
>     import org.apache.flink.api.common.functions.RichMapFunction;
>     import org.apache.flink.api.java.functions.KeySelector;
>     import org.apache.flink.cep.CEP;
>     import org.apache.flink.cep.PatternSelectFunction;
>     import org.apache.flink.cep.PatternStream;
>     import org.apache.flink.cep.RichPatternFlatSelectFunction;
>     import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
>     import org.apache.flink.cep.pattern.Pattern;
>     import org.apache.flink.cep.pattern.conditions.RichIterativeCondition;
>     import org.apache.flink.cep.pattern.conditions.SimpleCondition;
>     import org.apache.flink.configuration.Configuration;
>     import org.apache.flink.streaming.api.TimeCharacteristic;
>     import org.apache.flink.streaming.api.datastream.DataStreamSource;
>     import org.apache.flink.streaming.api.datastream.KeyedStream;
>     import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
>     import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
>     import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
>     import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
>     import org.apache.flink.streaming.api.windowing.time.Time;
>     import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
>     import org.apache.flink.util.Collector;
>     
>     import java.util.Iterator;
>     import java.util.List;
>     import java.util.Map;
>     import java.util.Properties;
>     
>     /**
>      * @Description: todo  实现业务场景：匹配紧邻数据时间差距小于1小时的数据结果
>     * @Author: greenday
>      * @Date: 2021/3/24 10:38
>      * <p>
>      * todo         com.sjb.test.yarn.CepOnYarnTestDemo1
>     */
>     public class CepOnYarnTestDemo2021_3_31_002 {
>     
>         public static void main(String[] args) throws Exception {
>     
>     
>             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
>             env.setParallelism(1);
>             env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>             env.getConfig().setAutoWatermarkInterval(1L);
>     
>     //        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
>     
>     Properties consumerProperties = getConsumerProperties();
>             FlinkKafkaConsumer<JSONObject> kafkaConsumer = new FlinkKafkaConsumer<JSONObject>(
>                     "flinkcep",
>                     new CustomerDeserializationSchema(),
>                     consumerProperties);
>     
>             //todo 策略模式
>     kafkaConsumer.setStartFromEarliest();
>             DataStreamSource<JSONObject> sourceStream = env.addSource(kafkaConsumer);
>     
>             KeyedStream<JSONObject, String> watermarkStream = sourceStream
>                     .assignTimestampsAndWatermarks( new BoundedOutOfOrdernessTimestampExtractor<JSONObject>(Time.milliseconds(1L)) {
>     
>                                 @Override
>     public long extractTimestamp(JSONObject element) {
>                                     return element.getLong("create_date");
>                                 }
>                             } ).keyBy(new KeySelector<JSONObject, String>() {
>                         @Override
>     public String getKey(JSONObject dataJson) throws Exception {
>                             return dataJson.getString("device_code");
>                         }
>                     });
>     
>     
>             Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("begin",AfterMatchSkipStrategy.skipToFirst("begin"))
>                     .where(new RichIterativeCondition<JSONObject>() {
>                         @Override
>     public boolean filter(JSONObject dataJson, Context<JSONObject> context) throws Exception {
>                             Integer operation_type = dataJson.getInteger("operation_type");
>                             if (operation_type == 1) {
>                                 return true;
>                             }
>                             return false;
>                         }
>                     }).next("middle")
>                     .where(new RichIterativeCondition<JSONObject>() {
>                         @Override
>     public boolean filter(JSONObject dataJson, Context<JSONObject> context) throws Exception {
>                             JSONObject begin = context.getEventsForPattern("begin").iterator().next();
>                             //todo 这里中间如果不能是 operation_type=24
>     Integer operation_type = dataJson.getInteger("operation_type");
>                             Long middleDate = dataJson.getLong("create_date");
>                             Long beginDate = begin.getLong("create_date");
>     
>                             if (operation_type == 1) {
>                                 if (middleDate - beginDate < 1100000L ){
>                                     return true;
>                                 }
>     
>                             }
>                             return false;
>                         }
>                     }).timesOrMore(2).consecutive().within(Time.minutes(5));
>     
>             //todo 规则
>     
>     PatternStream patternStream = CEP.pattern(watermarkStream, pattern);
>             SingleOutputStreamOperator select = patternStream.flatSelect(new RichPatternFlatSelectFunction<JSONObject,  List<JSONObject>>() {
>                 @Override
>     public void open(Configuration parameters) throws Exception {
>                     super.open(parameters);
>                 }
>     
>     
>                 @Override
>     public void flatSelect(Map<String, List<JSONObject>> map, Collector<List<JSONObject>> out) throws Exception {
>     
>                     JSONObject begin = map.get("begin").get(0);
>                     List<JSONObject> middle = map.get("middle");
>                     middle.add(0, begin);
>                     out.collect(middle);
>     
>     
>                 }
>     
>             });
>     
>     
>             select.print("结果输出--->");
>     
>     
>     //        select.addSink(new HBaseWriter());
>     env.execute("cep");
>         }
>     
>         public static Properties getConsumerProperties() {
>             Properties properties = new Properties();
>             properties.put("bootstrap.servers", Constants.BOOTSTRAP_SERVERS);
>             properties.put("group.id", Constants.GROUP_ID_TEST);
>             properties.setProperty("flink.partition-discovery.interval-millis", 300 * 1000 + "");  // 自动发现消费的partition变化
>     return properties;
>         }
>     
>     
>     }

执行结果：

>  结果输出--->> \[\{"operation\_type":"1","order\_number":"1","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653174000"\}, \{"operation\_type":"1","order\_number":"2","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653274000"\}, \{"operation\_type":"1","order\_number":"3","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653374000"\}\]  
> 结果输出--->> \[\{"operation\_type":"1","order\_number":"1","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653174000"\}, \{"operation\_type":"1","order\_number":"2","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653274000"\}, \{"operation\_type":"1","order\_number":"3","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653374000"\}, \{"operation\_type":"1","order\_number":"4","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653474000"\}\]  
> 结果输出--->> \[\{"operation\_type":"1","order\_number":"2","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653274000"\}, \{"operation\_type":"1","order\_number":"3","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653374000"\}, \{"operation\_type":"1","order\_number":"4","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653474000"\}\]

# 疑惑点：关于水印的环节，最后一条数据不进入cep触发计算。  #

**数据：**

**案例代码：**

>     DataStreamSource<JSONObject> sourceStream2 = env.addSource(new RichSourceFunction<JSONObject>() {
>                 @Override
>                 public void run(SourceContext<JSONObject> ctx) throws Exception {
>     
>                     ctx.collect(JSON.parseObject("{\"operation_type\":\"1\",\"order_number\":\"1\",\"device_code\":\"6f253e518373bcbe\",\"create_date\":\"1616653174000\"}"));
>                     ctx.collect(JSON.parseObject("{\"operation_type\":\"1\",\"order_number\":\"2\",\"device_code\":\"6f253e518373bcbe\",\"create_date\":\"1616653274000\"}"));
>                     ctx.collect(JSON.parseObject("{\"operation_type\":\"1\",\"order_number\":\"3\",\"device_code\":\"6f253e518373bcbe\",\"create_date\":\"1616653374000\"}"));
>                     ctx.collect(JSON.parseObject("{\"operation_type\":\"1\",\"order_number\":\"4\",\"device_code\":\"6f253e518373bcbe\",\"create_date\":\"1616653474000\"}"));
>                     while (true){
>                         Thread.sleep(10000);
>                     }
>     
>                 }
>     
>                 @Override
>                 public void cancel() {
>     
>                 }
>             });
>     
>             KeyedStream<JSONObject, String> watermarkStream = sourceStream2
>                     .assignTimestampsAndWatermarks( new BoundedOutOfOrdernessTimestampExtractor<JSONObject>(Time.milliseconds(1L)) {
>                                 @Override
>                                 public long extractTimestamp(JSONObject element) {
>                                     return element.getLong("create_date");
>                                 }
>                             } ).keyBy(new KeySelector<JSONObject, String>() {
>                         @Override
>                         public String getKey(JSONObject dataJson) throws Exception {
>                             return dataJson.getString("device_code");
>                         }
>                     });
>             Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("begin",AfterMatchSkipStrategy.skipToFirst("begin"))
>                     .where(new RichIterativeCondition<JSONObject>() {
>                         @Override
>                         public boolean filter(JSONObject dataJson, Context<JSONObject> context) throws Exception {
>                             Integer operation_type = dataJson.getInteger("operation_type");
>                             if (operation_type == 1) {
>                                 return true;
>                             }
>                             return false;
>                         }
>                     });
>     
>             //todo 规则
>     
>             PatternStream patternStream = CEP.pattern(watermarkStream, pattern);
>             SingleOutputStreamOperator select = patternStream.flatSelect(new RichPatternFlatSelectFunction<JSONObject,  List<JSONObject>>() {
>                 @Override
>                 public void open(Configuration parameters) throws Exception {
>                     super.open(parameters);
>                 }
>                 @Override
>                 public void flatSelect(Map<String, List<JSONObject>> map, Collector<List<JSONObject>> out) throws Exception {
>     
>                     List<JSONObject> begin = map.get("begin");
>     
>                     out.collect(begin);
>     
>     
>                 }
>     
>             });
>     
>     
>             select.print("结果输出--->");
>     
>     
>     //        select.addSink(new HBaseWriter());
>             env.execute("cep");

执行结果：

>  结果输出--->> \[\{"operation\_type":"1","order\_number":"1","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653174000"\}\]  
> 结果输出--->> \[\{"operation\_type":"1","order\_number":"2","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653274000"\}\]  
> 结果输出--->> \[\{"operation\_type":"1","order\_number":"3","device\_code":"6f253e518373bcbe","kafka\_topic":"flinkcep","create\_date":"1616653374000"\}\]

你看懂了吗？

验证之后先不管它了，只要kafka有数据进来，之前的数据就会触发，触发kafka一直没数据进来，最后一条数据才不会被触发计算 ，底层原因就是因为kafka的偏移量没有变化，Flink底层是这么设计的，有大佬说试试了解一下idle解决，验证了1.12版本的空闲水印，也是一样，不是解决这个最后一条数据触发问题，而且0.1.12版本基础代码API也有变化，替换成本有点高。目前是我知道的这些情况，先搁置吧。