考虑到我对您的要求是正确的，我将作为一个 FlatMapFunction 使用fifo队列。队列缓冲区 k 事件并在新事件到达时发射头部。如果需要容错流应用程序，则必须将队列注册为状态。flink将负责检查状态（即队列），并在出现故障时恢复状态。
这个 FlatMapFunction 可能是这样的：

class Lagger(val k: Int) 
    extends FlatMapFunction[X, X] 
    with Checkpointed[mutable.Queue[X]] 
{

  var fifo: mutable.Queue[X] = new mutable.Queue[X]()

  override def flatMap(value: X, out: Collector[X]): Unit = {
    // add new element to queue
    fifo.enqueue(value)
    if (fifo.size == k + 1) {
      // remove head element and emit
      out.collect(fifo.dequeue())
    }
  }

  // restore state
  override def restoreState(state: mutable.Queue[X]) = { fifo = state }

  // get state to checkpoint
  override def snapshotState(cId: Long, cTS: Long): mutable.Queue[X] = fifo

}

返回具有时滞的元素更为复杂。这将需要用于发射的计时器线程，因为只有在新元素到达时才调用该函数。