Apache Flink：创建滞后数据流

Question

val ds : DataStream = ...

val laggedDS : DataStream = ds.map(lag _)

def lag(ds : DataStream, k : Time) : DataStream = {

}

如果每个事件间隔为1秒，并且有2秒的滞后，那么我希望示例输入和输出如下所示。

输入：1,2,3,4,5,6,7...
输出：NA,NA,1,2,3,4,5...

Answer 1

如果您的需求正确，我将把它实现为具有FIFO队列的FlatMapFunction。队列缓冲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

}

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