Apache Flink : 지연된 Datastream 만들기

Question

나는 Scala를 사용하여 Apache Flink를 시작했습니다. 누군가 내가 가지고있는 현재의 데이터 스트림에서 느린 스트림 (k 이벤트 또는 k 단위의 시간 지연)을 만드는 방법을 알려주시겠습니까?

기본적으로 데이터 스트림에서 자동 회귀 모델 (시간 지연 버전의 선형 선형 회귀 모델)을 구현하고 싶습니다. 따라서 다음의 의사 코드와 비슷한 방법이 필요합니다. 모든 이벤트를 1 개 초 간격으로 이격 된 경우

val ds : DataStream = ... 

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

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

} 

은이 같은 샘플의 입력 및 출력을 기대하고 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 

} 

돌아 요소가 더 복잡하다. 새 요소가 도착할 때만 함수가 호출되기 때문에 방출을위한 타이머 스레드가 필요합니다.