스파크 스트리밍 쿼리 101

몇 년 동안이나 pyspark로 대부분의 데이터 처리를 해왔지만 스트리밍은 처음이라 기초적인 내용들 정리해봄

• 구조적 스트리밍

  • 스트림 데이터를 무한히 증가하는 unbounded 테이블의 일부로 새로운 행이 추가되는 것으로 간주
  • 소스 = 데이터가 들어오는 곳(e.g. 카프카), 싱크 = 데이터를 쓰는 목적지
  • 기본적으로는 배치 처리할 때와 크게 다르지 않은 방식(sparksql)으로 연산(transformation, action)이 가능함

• 연산(Operations)

  • 기본적인 SQL-like operations 적용 가능(select, where, groupBy)

• aggregation시 거의 항상 time-based window 사용

  • unbounded table 개념 생각해보면 단순 groupby를 치면 데이터가 끝없이 쌓이므로 state를 유지하기 불가능함
  • N분/시간/일 단위 등 시간으로 잘라서 상태를 제한적으로 유지해서 집계

• 가능한 time-based window는 3가지 존재
  • tumbling: 겹치지 않는(disjoint) 고정 길이의 시간 구간으로 데이터를 나누는 방식
  • sliding: 고정 길이의 윈도우를 일정 간격(슬라이드 간격)으로 계속 이동시키며 집계하는 방식
    • tumbling과 달리 각 이벤트가 여러 윈도우에 속할 수 있음
  • session: 이벤트 간의 유휴 시간(gap)을 기준으로 윈도우를 나누는 방식
    • 윈도우 길이가 고정되어 있지 않고 지정한 갭보다 이벤트가 없을 때 세션을 종료함

df.groupBy(F.window("event_time", "10 minutes")).agg(F.count("*").alias("cnt")) #tumbling
df.groupBy(F.window("event_time", "10 minutes", "5 minutes")).agg(F.count("*")) #sliding
df.groupBy(F.session_window("event_time", "10 minutes"), "user_id").agg(F.count("*")) #session

• watermark
  • 데이터가 들어오는 시간
    • 이벤트 시간 (데이터 안에 존재하는 시간 칼럼)
    • 수집 시간 (데이터가 스트림 처리 엔진으로 수집된 시간)
    • 처리 시간 (스트림 처리 엔진이 데이터를 처리하는 시간)
  • 늦게 도착하는 데이터는 어떻게 하나?
    • watermark를 설정한 만큼만 기다려 주겠다는 의미 = 진짜 마지막 데드라인

df.withWatermark(time_column, threshold)
#e.g. df.withWatermark("event_time", "5 minutes")

• 위의 예시

  • 12:04(event time)에 발생한 dog는 사실 12:00-12:10 window에 들어가야 함
  • 만약 watermark를 5분까지로 설정하면 12:15까지는 해당 window를 열어두고 기다림
  • watermark 내에도 도착하지 못한 데이터는 그냥 버리고, 12:00-12:10 window 집계도 state에서 삭제하고 다음 걸로 넘어감

• output mode

  • append mode: 지난 트리거 이후로 새로 발생한 데이터만 출력
  • update mode: 새로 들어온 정보로 기존의 데이터를 갱신
  • complete mode: 매번 전체 결과를 덮어씀

• 🚨스트리밍에서는 안 되는 쿼리 (operations) 🚨

  • 밥먹듯이 쓰던 게 스트리밍에서는 안되는 경우가 많았다 예를 들면,
    • limit 및 처음 N개의 행 가져오기(take) 연산
    • distinct 연산 -> 해보니 approx count distinct로 대체할 것을 권장한다! 매우 정확해야 하는 상황이 아니면 대안이 될 수 있음
    • sort는 집계 이후에만 지원되고 complete output mode에서만 가능
    • 일부 outer join (아래 support matrix 참고)
    • event time or processing time 컬럼을 기반으로 하는 time-based window가 아닌 다른 Window 연산 - 즉 일반적으로 많이 사용하는 row_number, rank, lead/lag 등등
    • pivot
    • 즉시 결과를 요구하는 액션들 (collection, toPandas, count, foreach, show 등)

static이냐 stream이냐 누가 왼쪽이냐에 따라 지원여부 달라짐. steam-stream join은 제약이 강함..

https://spark.apache.org/docs/latest/streaming/index.html