来自Flink训练教程的一个问题：longridessolution.scala

Question

override def processElement(ride: TaxiRide,
                                context: KeyedProcessFunction[Long, TaxiRide, TaxiRide]#Context,
                                out: Collector[TaxiRide]): Unit = {
      val timerService = context.timerService
      if (ride.isStart) {
        // the matching END might have arrived first; don't overwrite it
        if (rideState.value() == null) {
          rideState.update(ride)
        }
      }
      else {
        rideState.update(ride)
      }

      timerService.registerEventTimeTimer(ride.getEventTime + 120 * 60 * 1000)
    }

一旦水印到达时间戳，计时器就会触发

    override def onTimer(timestamp: Long,
                         ctx: KeyedProcessFunction[Long, TaxiRide, TaxiRide]#OnTimerContext,
                         out: Collector[TaxiRide]): Unit = {
      val savedRide = rideState.value

      if (savedRide != null && savedRide.isStart) {
        out.collect(savedRide)
      }
      rideState.clear()
    }

问题是：如果结束记录在先，然后基于逻辑，它不会更新骑行状态（相关键），然后2小时后触发，然后它不会收集，不会发出记录，但是如果这个记录符合我们的要求呢？==>记录开始时间发生在2个多小时前？我认为应该有更多的逻辑来处理这件事

Answer 1

如果结束记录是在开始记录之前处理的，那么开始记录可能很晚才到达，当它到达时，它提供了这段旅程持续了两个多小时的证据。

但是，这个练习的目标不是找到所有持续时间超过两个小时的游乐项目，而是实时地标记那些现在应该结束的游乐项目（因为它们在两个多小时前就开始了），但还没有结束。由于你询问的这些游乐项目已经结束，是否值得提醒是有争议的。

您提出了一个有趣的观点，可能应该添加到练习讨论页面中。