[Spark] 스파크 스트리밍의 이벤트 시간 처리와 Watermark

스트리밍 쿼리에서의 집계는 보통 현재시간부터 T시간 전까지 집계된 데이터를 의미한다.

그런데 예를들어 지난 15초 동안의 집계 데이터라고 하면 15초의 시간은 언제부터 언제까지인지 의문이 생길 수 있다.

스트림 처리에서 바라보는 시간은 다양한 기준이 있다.

스트림 처리에서의 시간 관점

• 이벤트 시간 (event time)
  데이터 의존적인 타임스템프로, 데이터 내에 존재하는 데이터 발생 시간이다.
  데이터 의존적이기 때문에 어떠한 값을 타임스탬프로 넣을 것인지는 다르겠지만 주로 이벤트가 발생한 시간을 많이 사용한다.
  windowing 에 사용할 수 있는 아주 적절한 시간 중 하나이다.
• 수집 시간 (ingestion time)
  스트림 처리 엔진에 데이터가 처음으로 수집된, 즉 들어온 시간을 나타낸다.
• 처리 시간 (processing time)
  실제 스트림 처리 엔진에서 데이터를 처리하는 시간이다. 즉, 해당 데이터를 받아 처리할 때 처리하는 서버의 시간이다.

비슷해 보이지만 아래의 이미지에서 명확하게 구분할 수 있다.

https://medium.com/big-data-processing/time-attributes-in-apache-flink-85e2afdda238

 

이벤트 기반 스트림 처리에서의 Challenges

이벤트 기반 스트림 처리에서의 문제는 아래의 내용이 있다.

• late event (지연 이벤트)
  소스에서 발생한 이벤트가 네트워크 지연이나 시스템 중단으로 인해 처리 엔진에 늦게 도착될 수 있다.
  스파크는 지난 15초동안의 이벤트 시간 창의 모든 이벤트가 처리 시스템에 도달했는지 알 수 있어야 한다.
  스파크는 기본적으로 지연 이벤트를 무기한 기다린다.
• state management
  지연 이벤트가 도착한 경우 (예를들어) 15초의 특정 이벤트 시간 window에 대해 스파크는 모든 이벤트를 대조하고 집계를 적용한다. 기본적으로 지연 이벤트를 수용하기 위해서 스파크는 window의 상태를 메모리에 기억해야 한다.
  시간이 지남에 따라 window의 수는 늘어날 것이고, 리소스 사용량도 증가할 것이다.
• state recovery
  스트리밍 작업이 실패한 경우 내결함성 보장을 위해 상태 복구를 할 수 있어야 한다.
  이부분은 다른 게시글에 정리했다.

 

지연 이벤트의 처리

네트워크 레이턴시 등의 잉유로 이벤트가 어플리케이션에 늦게 도착할 수 있다.

아래의 이미지와 같이 예를들어 12:04에 생성된 dog가 12:11에 스파크 스트리밍에 수신되는 경우에는 [12:00, 12:10) 윈도우에 업데이트를 해야한다.

사실 스파크에서 지연 이벤트를 무기한 기다리며 상태를 무제한 저장하는 것은 한정된 리소스 차원에서 불가능하다.

일정 임계값(threshold)을 정해두고 임계값을 넘어가는 이벤트는 버릴 수 있어야한다. 즉 이전 집게가 메모리 내 state에서 삭제될 수 있는 시기를 시스템이 알아야 한다는 의미이다.

이렇게 제한된 방식으로 상태를 제어하는 메커니즘인 watermark를 사용할 수 있다.

 

 

watermark란

워터마크는 지연 이벤트를 수용하기 위해 window의 state를 관리하는 기법이다. state가 무한정 커지지 못하도록 제어한다.

늦은 임계값에 대한 값을 정의하고, 임계값을 통과하면 만료된 window의 상태를 삭제하고 만료된 window에 해당하는 지연 이벤트를 쿼리 처리에 반영하지 않는다.

워터마크는 `(max event time seen by the engine - late threshold > T)` 인 T 시간에 해당하는 지연 이벤트만 처리하도록 허용한다.

• 도입시기: Spark 2.1.0

 

Update Output Mode에서의 Watermark

워터마크에서는 아래와 같이 프로세싱 전/후에 값을 업데이트 한다.

• before processing
  watermark 기반으로 batch to finalize를 업데이트 한다.
  워터마크 값보다 배치의 end of boundary 중 가장 작은 값이 작은 경우에 업데이트되며,
  더 이상 해당하는 배치의 데이터에는 변경이 없다는 의미이다.
• after trigger
  microbatch processing 이후에 watermark 값이 업데이트 된다.
  watermark 값은 오래된 데이터의 limit 을 정하는 일종의 barrier 역할을 한다.

위 wordcount 이미지를 통해 이해할 수 있다.

• batch interval: 10분
• trigger interval: 5분
• watermark: 10분

시간별 watermark의 변화는 아래와 같다.

 

지연 이벤트 확인

• 12:09, cat (12:15 triggered)
  이때는 watermark 가 아직 미설정되어있으므로 집계에 사용된다.
  [12:00, 12:10) 윈도우에 업데이트된다.
• 12:08, dog / 12:13, owl (12:20 triggered)
  12:20 시점의 before processing에서 watermark는 12:04이다,
  dog와 owl 모두 12:04보다 큰 값이므로 업데이트된다.
• 12:04, donkey (12:25 triggered)
  12:25시점의 before processing에서 watermark는 12:11이다.
  donkey의 event time인 12:04는 watermark보다 작은 값이므로 dropped된다.
• 12:17, owl (12:25 triggered)
  12:25시점의 before processing에서 watermark는 12:11이다.
  12:17은 watermark보다 큰 값이므로 집계된다.

추가로 12:21, owl은 시간 흐름으로 봤을 때 12:20 이전에 event time이 12:21 로 수신된 것을 확인할 수 있다.

소스와 처리엔진의 clock이 동기화되어 있지 않거나 여러 device에서 generate 되는 경우 시간 동기화 문제로 일어날 수 있는 흔한 이슈이다.

 

batches to finalize

12:25 시점 before processing에서 batches to finalize가 설정된 것을 볼 수 있다.

12:25 before processing 때의 워터마크는 12:11이고, 워터마크보다 작은 배치가 생겼으므로 그 배치에 해당하는 데이터들은 더이상 업데이트가 되지 않는다.

 

Append Output Mode에서의 Watermark

update 때와 비슷하지만 차이점은 아래와 같다.

• batch가 finalized 되기 전까지 모든 데이터는 internal state에 저장되어 있다.

 

아래의 표에서 볼 수 있다시피 batches to finalized가 12:25 와 12:30 시점에서 정해지기 때문에(업데이트되기 때문에),

result는 12:25 시점부터 sink하게 된다.

 

watermark 사용 방법

withWatermark를 사용한다.

`DataFrame.withWatermark(eventTime, delayThreshold)`

• eventTime: str or column
  행의 이벤트 시간을 포함하는 column 이름이다.
• delayThreshold: str
  간격 형식으로 처리된, 최신 레코드를 기준으로 데이터가 늦게 도착할 때까지 기다리는 최소 지연시간이다.

windowedCounts = words \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(
        window(words.timestamp, "10 minutes", "5 minutes"),
        words.word) \
    .count()

 

update mode 에서는 stateful 연산 시 (집계/join등)에는 선택적으로 사용 가능하며,

append mode 에서 stateful 연산 사용 시에는 (집계/join등) 필수로 watermark값을 정의해주어야 한다.

 

참고

https://seamless.tistory.com/99

https://medium.com/big-data-processing/time-attributes-in-apache-flink-85e2afdda238

https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#handling-late-data-and-watermarking

https://www.youtube.com/watch?v=XjlKGvUt2dY

https://spark.apache.org/docs/3.1.1/api/python/reference/api/pyspark.sql.DataFrame.withWatermark.html

https://stackoverflow.com/questions/44403690/empty-output-for-watermarked-aggregation-query-in-append-mode