flink数据不由timewindow运算符中的process函数处理

Question

我有一个时间窗口，我尝试确定我是否在一段时间内获得一个新的密钥。我正在通过kafka推送数据，当我调试它时，我看到数据到达keyby方法，但它没有到达process方法，并且没有被收集器收集。我正在使用BoundedouToFordernesTimeStampExtractor来分配水印：

    case class Src(qip:Ip, ref: Ip, ts: Long) extends FooRequest

    class TsExtractor extends BoundedOutOfOrdernessTimestampExtractor[Src](Time.hours(3)){
      override def extractTimestamp(element: Src): Long = element.ts
    }

    class RefFilter extends ProcessWindowFunction[Src, IpDetectionSrc, String, TimeWindow]{
      private lazy val stateDescriptor = new ValueStateDescriptor("refFilter",  createTypeInformation[String])

      override def process(key: String, context: Context, elements: Iterable[Src], out: Collector[IpDetectionSrc]): Unit = {
        println(s"RefIpFilter processing $key")//data is not getting here 
        if(Option(context.windowState.getState(stateDescriptor).value()).isEmpty){
          println(s"new key found $key") //data is not getting here also 
          context.windowState.getState(stateDescriptor).update(key)
          out.collect(elements.head)
        }
      }
    }

lazy val env: StreamExecutionEnvironment =
    setupEnv(StreamExecutionEnvironment.getExecutionEnvironment)(300000,Some(stateDir), Some(TimeCharacteristic.EventTime))

 lazy val src: DataStream[FooRequest] = env.addSource(consumer)

 lazy val uniqueRef:DataStream[FooRequest] => DataStream[Src] = src => src 
        .flatMap(new FlatMapFunction[FooRequest,Src ]{
          override def flatMap(value: FooRequest, out: Collector[Src]): Unit =   value match {
            case r: Src =>
              out.collect(r)
            case invalid =>
              log.warn(s"filtered unexpected request $invalid")
          }
        })
        .assignTimestampsAndWatermarks(new TsExtractor)
        .keyBy(r => r.ref)
        .timeWindow(Time.seconds(120))
        .allowedLateness(Time.seconds(360))
        .process(new RefFilter)

uniqueRef(src).addSink(sink)
env.execute()

如有任何协助，我将不胜感激

Answer 1

BoundedouToFordernesTimeStampExtractor跟踪到目前为止看到的最高时间戳，并根据配置的延迟（在本例中为3小时）生成水印。这些水印是定期产生的，默认情况下每200毫秒产生一次。所以只有一个单一的事件，水印将落后于这个事件3小时，窗口永远不会被触发。此外，在输入有限的情况下，作业在处理完所有事件后将停止运行。

context.windowstate是每个窗口的状态，具有有限的生存期。每个2分钟窗口将有自己的实例，一旦窗口的允许迟到时间过期，它就会被清除。如果您希望具有全局作用域的键控窗口状态具有不确定的生存期，请改用context.globalstate。