代码解析

定义输入输出以及构造函数

  1.     private final SourceFunction<TaxiRide> source;
  2.     private final SinkFunction<Long> sink;
  4.     public LongRidesExercise(SourceFunction<TaxiRide> source, SinkFunction<Long> sink){
  5.         this.source = source;
  6.         this.sink = sink;
  7.     }

execute()

  • 定义流环境以及输入
  • 定义watermarkStrategy
  • keyBy处理，process

  • addSink

    1. public JobExecutionResult execute() throws Exception {
    2.       StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    3.       DataStream<TaxiRide> rides = env.addSource(source);
    5.       // TODO，自行定义与处理watermarkStrategy，此处用途是什么？
    6.       WatermarkStrategy<TaxiRide> watermarkStrategy =
    7.               WatermarkStrategy.<TaxiRide>forBoundedOutOfOrderness(Duration.ofSeconds(60))
    8.                       .withTimestampAssigner((ride,streamRecordTimestamp) -> ride.getEventTimeMillis());
    9.       rides.assignTimestampsAndWatermarks(watermarkStrategy)
    10.               .keyBy(ride -> ride.rideId)
    11.               .process(new AlertFunction())
    12.               .addSink(sink);
    14.       return env.execute();
    15.   }

    AlertFunction

  • 继承抽象类KeyedProcessFunction

  • 分别实现open、processElement、onTimer 3个方法
  • open函数处理状态
  • processElement函数处理业务逻辑，包括判断状态、生成计时器，处理输出流、清理状态等
  • onTimer函数处理触发到计时器后的逻辑
  • getTimerTime和RideTooLong是辅助函数
  1.     public static class AlertFunction extends KeyedProcessFunction<Long,TaxiRide,Long>{
  2.         private ValueState<TaxiRide> rideState;
  3.         @Override
  4.         public void open(Configuration config){
  5.             ValueStateDescriptor<TaxiRide> rideStateDescriptor =
  6.                     new ValueStateDescriptor<>("ride event", TaxiRide.class);
  7.             rideState = getRuntimeContext().getState(rideStateDescriptor);
  8.         }
  10.         @Override
  11.         public void processElement(TaxiRide taxiRide, Context context, Collector<Long> out) throws Exception {
  12.             TaxiRide firstRideEvent = rideState.value();
  14.             if(firstRideEvent == null){
  15.                 //无论是开始事件和结束事件谁先到来，都记录到状态中
  16.                 rideState.update(taxiRide);
  18.                 // 如果是开始事件，生成一个计时器
  19.                 if(taxiRide.isStart){
  20.                     context.timerService().registerEventTimeTimer(getTimerTime(taxiRide));
  21.                 }
  22.             }else{
  23.                 if(taxiRide.isStart){
  24.                     if (rideTooLong(taxiRide, firstRideEvent)) {
  25.                         out.collect(taxiRide.rideId);
  26.                     }
  27.                 }else{
  28.                     //状态中存储了start事件，计时器仍在运行，除非它被触发了。此处删除计时器
  29.                     context.timerService().deleteEventTimeTimer(getTimerTime(firstRideEvent));
  31.                     // 有可能没有触发计时器报警，再次判断下输出
  32.                     if(rideTooLong(firstRideEvent,taxiRide)){
  33.                         out.collect(taxiRide.rideId);
  34.                     }
  35.                 }
  37.                 //因为两个事件都出现了，所以可以清空状态
  38.                 //当一个事件丢失时，本处理方法可能会导致内存泄漏
  39.                 // TODO，后面我们更多讨论
  40.                 rideState.clear();
  42.             }
  43.         }
  45.         @Override
  46.         public void onTimer(long timestamp, OnTimerContext context, Collector<Long> out) throws IOException {
  47.             // 超时行程，触发计时器
  48.             out.collect(rideState.value().rideId);
  50.             // 清空状态，防止再次触发。但是会引发一个新问题，当end事件到达时，会漏掉之前的开始状态
  51.            rideState.clear();
  52.         }
  54.         private long getTimerTime(TaxiRide ride) throws RuntimeException {
  55.             if (ride.isStart) {
  56.                 return ride.eventTime.plusSeconds(120 * 60).toEpochMilli();
  57.             } else {
  58.                 throw new RuntimeException("Can not get start time from END event.");
  59.             }
  60.         }
  62.         private boolean rideTooLong(TaxiRide startEvent, TaxiRide endEvent) {
  63.             return Duration.between(startEvent.eventTime, endEvent.eventTime)
  64.                     .compareTo(Duration.ofHours(2))
  65.                     > 0;
  66.         }
  67.     }