Spark

정적 DataFrame으로 먼저 로직을 검증하기

예제로 쓰이는 소매 데이터는 하루 단위 CSV 파일들로 나뉘어 있고, 실제 상황이라면 매장에서 지속적으로 생성되어 특정 위치로 전송된 뒤 Structured Streaming 잡이 읽어가는 형태를 흉내 낸 것이다. 스트리밍으로 바로 들어가기 전에, 먼저 이 데이터를 정적(static) DataFrame으로 읽어 로직을 검증한다. 이때 스키마를 한 번 추론해 staticSchema에 저장해 두면, 이후 스트리밍으로 같은 데이터를 읽을 때 스키마 추론을 반복하지 않아도 된다.

예제 쿼리는 각 고객(CustomerId)이 하루 동안 가장 많이 지출한 시점을 찾는 것이다. UnitPrice와 Quantity를 곱해 total_cost 컬럼을 만들고, CustomerId와 window(InvoiceDate, "1 day")로 그룹핑한 뒤 total_cost 합계를 내림차순 정렬해 상위 5건을 가져온다.

staticDataFrame
  .selectExpr("CustomerId", "(UnitPrice * Quantity) as total_cost", "InvoiceDate")
  .groupBy(col("CustomerId"), window(col("InvoiceDate"), "1 day"))
  .sum("total_cost")
  .orderBy(desc("sum(total_cost)"))
  .take(5)

read를 readStream으로 바꾸기만 하면 되는 이유

정적 버전과 스트리밍 버전의 코드 차이는 놀라울 만큼 작다. 가장 큰 변화는 read 대신 readStream을 쓴다는 것과, maxFilesPerTrigger 옵션을 추가한다는 것뿐이다. 이 옵션은 한 트리거마다 읽어들일 파일 개수를 제한하는데, 이는 데모에서 스트리밍처럼 보이도록 만들기 위한 장치이고 실제 운영 환경에서는 보통 생략한다.

로컬 모드로 실습할 때는 spark.sql.shuffle.partitions 설정도 함께 조정하는 것이 좋다. 이 값은 셔플 이후 생성되는 파티션 수를 지정하며 기본값은 200인데, 로컬 머신에는 익스큐터가 몇 개 없으므로 5 정도로 줄이는 것이 적절하다.

streamingDataFrame.isStreaming을 호출하면 true가 반환되어 이 DataFrame이 스트리밍 소스에서 왔음을 확인할 수 있다. 다만 이 시점에도 여전히 지연 실행(lazy) 상태이므로, 실제 실행을 시작하려면 writeStream으로 스트리밍 액션을 호출해야 한다.

spark.conf.set("spark.sql.shuffle.partitions", "5")

val streamingDataFrame = spark.readStream
  .schema(staticSchema)
  .option("maxFilesPerTrigger", 1)
  .format("csv")
  .option("header", "true")
  .load("dbfs:/mnt/defg/retail-data/by-day/*.csv")

streamingDataFrame.isStreaming // true

writeStream, outputMode, 그리고 이벤트 타임

스트리밍 쿼리를 시작하려면 writeStream을 호출한다. 예제는 format("memory")로 in-memory 테이블에 결과를 쓰고, queryName("customer_purchases")로 테이블 이름을 지정하며, outputMode("complete")로 매 트리거마다 전체 결과 테이블(모든 집계 값)을 다시 계산해 반영하도록 한다. 이렇게 시작한 쿼리는 이후 spark.sql로 customer_purchases 테이블을 조회하면 데이터가 더 들어올수록 상위 고객의 구매액이 갱신되는 것을 관찰할 수 있다. format("console")로 콘솔에 결과를 출력하는 방법도 있지만, memory와 console 두 sink 모두 데모·디버깅 용도일 뿐 운영 환경에서 사용할 것은 아니다.

여기서 짚어야 할 중요한 지점은, 이 윈도우 집계가 Spark가 데이터를 처리한 시각(처리 시간)이 아니라 데이터 자체에 기록된 InvoiceDate, 즉 이벤트 타임을 기준으로 만들어진다는 점이다. 처리 시간에만 의존하던 것은 기존 Spark Streaming(DStream 기반)의 한계 중 하나였고, Structured Streaming은 이를 보완한다. Structured Streaming은 Spark 2.2부터 프로덕션 레디로 제공되며, 배치용으로 작성한 쿼리를 거의 그대로 스트리밍으로 전환할 수 있게 해준다는 것이 이 장이 강조하는 핵심 메시지다.

purchaseByCustomerPerHour.writeStream
  .format("memory")
  .queryName("customer_purchases")
  .outputMode("complete")
  .start()