Apache Spark — 느린 잡의 답은 실행 계획과 이벤트 로그에 있다 · Structured Streaming · 讲解

스트리밍은 작은 배치의 연속이고 체크포인트가 기억을 맡는다

한 줄 요약

Structured Streaming 은 끝없이 늘어나는 표에 같은 쿼리를 작은 배치로 거듭 돌리는 엔진이고, 어디까지 처리했는지는 체크포인트가, 무엇을 써 냈는지는 싱크의 기록이 기억한다. 늦게 온 자료를 얼마나 기다릴지는 워터마크가 정한다.

概念图: 작은 배치로 거듭 돌리는 · 입력 표 · 결과 표 · 마이크로 배치

왜 배치를 매일 다시 돌리면 안 되나

착륙 폴더에 협력사 파일이 하루에도 수십 번 떨어진다고 하자. 배치 잡으로 풀면 두 길뿐이다. 매번 폴더 전체를 다시 읽거나(자료가 쌓일수록 느려진다), 이미 읽은 파일 목록을 직접 관리하거나(잡이 도중에 죽으면 목록과 결과가 어긋난다). 두 번째 길을 제대로 만들려면 결국 "처리하기로 한 것을 먼저 적고, 다 쓴 뒤 끝났다고 적는" 장치를 손으로 짜게 된다.

Structured Streaming 은 그 장치를 엔진에 넣어 둔 것이다. [시작하기 문서](https://spark.apache.org/docs/4.2.0/streaming/getting-started.html)는 들어오는 자료를 계속 행이 덧붙는 입력 표로 보고, 쿼리를 그 위의 결과 표로 본다. 트리거마다 새 행만 가지고 결과를 갱신한다. 쿼리는 배치와 똑같이 쓰고, 증분으로 돌리는 일은 엔진이 맡는다. [개요 문서](https://spark.apache.org/docs/4.2.0/streaming/index.html)에 따르면 기본 실행 방식은 이 일을 작은 배치 잡의 연속으로 처리하는 마이크로 배치다.

어떻게 동작하나 — 파일 원본

[API 문서](https://spark.apache.org/docs/4.2.0/streaming/apis-on-dataframes-and-datasets.html)의 파일 원본은 디렉터리에 새로 나타난 파일을 읽는다. 알아 둘 규칙이 넷 있다.

파일은 수정 시각 순서로 처리된다.
파일은 디렉터리에 원자적으로 놓여야 한다. 대부분의 파일 시스템에서는 다른 곳에 다 쓴 뒤 옮기는(move) 것으로 된다. 제자리에서 천천히 쓰면 반쯤 쓴 파일을 읽을 수 있다.
새 파일인지는 기본으로 전체 경로로 판별한다(fileNameOnly 기본 false). 내용이 같아도 이름이 다르면 새 파일이다.
maxFilesPerTrigger 는 한 트리거가 집을 새 파일 수의 상한이고, 기본은 제한이 없다. maxFileAge 는 기본 1주이지만 첫 배치에서는 모든 파일을 유효하게 본다.

세 번째 규칙이 실무의 중복을 만든다. 협력사가 어제 보낸 묶음을 이름만 바꿔 다시 보내면 엔진은 그것을 새 파일로 받아들인다. 파일 단위의 "한 번씩" 은 보장되지만 내용 단위의 한 번씩은 아니다. 그것은 뒤에서 중복 제거로 푼다.

체크포인트와 싱크의 기록

마이크로 배치 하나가 도는 순서를 그린 그림. 착륙 폴더에서 새 파일을 고른 뒤 체크포인트의 offsets 에 이번 배치가 처리할 범위를 먼저 적고, 처리한 결과를 출력 폴더에 파일로 쓴 뒤 _spark_metadata 에 그 파일 목록을 적고, 마지막으로 commits 에 끝났다고 적는다. 도중에 죽으면 commits 가 없는 배치를 같은 범위로 다시 돌린다

시작하기 문서의 장애 허용 절은 설계를 한 문장으로 요약한다. 원본마다 읽은 위치를 나타내는 오프셋이 있고, 엔진은 체크포인트와 선행 기록(write-ahead log) 으로 트리거마다 처리할 오프셋 범위를 기록한다. 싱크는 같은 배치를 다시 받아도 결과가 같도록(멱등하게) 설계되어 있다. 다시 읽을 수 있는 원본과 멱등한 싱크가 만나 끝에서 끝까지 정확히 한 번이 된다.

체크포인트 디렉터리를 열어 보면 이 설계가 파일로 보인다. offsets/0 은 0번 배치가 처리하기로 한 범위이고 처리 전에 쓰인다. commits/0 은 그 배치가 끝났다는 표시이고 처리 뒤에 쓰인다. 재시작한 쿼리는 offsets 에는 있는데 commits 에는 없는 배치를 찾아 같은 범위로 다시 돌린다. 파일 싱크 쪽에도 짝이 되는 기록이 있다. 출력 폴더의 _spark_metadata/0 에 0번 배치가 써 낸 파일 목록이 적히고, Spark 로 그 폴더를 읽으면 이 목록에 있는 파일만 본다. 죽은 배치가 남긴 반쪽 파일이 결과에 섞이지 않는 이유다. API 문서의 표에서 파일 싱크가 정확히 한 번으로 표시되는 것도 이 기록 덕이다.

그래서 체크포인트를 지우는 것은 기억을 지우는 것이다. 같은 쿼리를 새 체크포인트로 띄우면 폴더의 모든 파일을 처음부터 다시 처리한다. 같은 문서는 재시작 사이에 파일 싱크의 출력 경로나 중복 제거 칼럼을 바꾸는 것도 허용되지 않는 변경으로 든다.

트리거 — 언제 배치를 돌리나

트리거를 주지 않으면 앞 배치가 끝나는 대로 다음 배치를 돌린다. 간격을 주면 그 간격마다 돈다. 이 실습처럼 있는 것만 다 처리하고 멈추는 쓰임에는 availableNow 가 맞다. API 문서에 따르면 실행 시점에 있는 자료를 모두 처리한 뒤 스스로 멈추는데, 원본 옵션(파일 원본이면 maxFilesPerTrigger)에 따라 여러 배치로 나눠 처리하고, 앞선 실행에서 커밋되지 못한 배치를 먼저 처리한다. 예전의 한 번 트리거(once)는 폐기 예정이다.

q = (spark.readStream.schema(schema).json("/root/landing")       .withWatermark("event_time", "10 minutes")       .dropDuplicates(["event_id", "event_time"])       .writeStream.format("parquet")       .option("path", "/root/out")       .option("checkpointLocation", "/root/chk")       .trigger(availableNow=True)       .start())q.awaitTermination()

중복 제거와 워터마크

스트리밍의 dropDuplicates 는 배치와 뜻이 같지만 대가가 다르다. [dropDuplicates 문서](https://spark.apache.org/docs/4.2.0/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.dropDuplicates.html)에 따르면 스트리밍에서는 이미 본 키를 트리거를 넘어 모두 상태로 들고 있어야 한다. 워터마크가 없으면 그 상태는 끝없이 자란다.

워터마크는 "이보다 늦게 오는 자료는 더 기다리지 않는다" 는 선이다. [withWatermark 문서](https://spark.apache.org/docs/4.2.0/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.withWatermark.html)는 그 선을 지금까지 본 최대 이벤트 시각에서 임계값을 뺀 값으로 정의한다. 이 선이 두 가지 일을 한다. 창 집계의 어느 창이 확정되었는지 알려 주고, 확정된 창의 상태를 치운다. 그래서 Append 모드의 창 집계는 창이 끝났다고 곧바로 나오지 않는다. 워터마크가 창의 끝을 넘은 뒤에야 한 번 나온다.

보장은 한 방향뿐이라는 점을 꼭 기억한다. API 문서는 10분 워터마크가 10분 안쪽으로 늦은 자료는 절대 버리지 않는다고 보장하지만, 그보다 늦은 자료가 반드시 버려진다고는 말하지 않는다. 대개 버려지지만 집계될 수도 있다. 워터마크는 늦은 자료를 거르는 필터가 아니라 상태를 치우는 기준이다. 그리고 집계에서 상태를 치우려면 워터마크를 집계 앞에, 집계에 쓰는 같은 시각 칼럼에 걸어야 하고 출력 모드는 Append 나 Update 여야 한다.

현장에서 만나는 모습

첫째, 체크포인트를 임시 폴더에 둔다. 재부팅 한 번에 기억이 사라지고 쿼리가 모든 파일을 다시 처리한다. 체크포인트는 출력만큼 소중한 자료다.

둘째, 착륙 폴더에 제자리 쓰기를 한다. 업로드 도구가 파일을 쓰는 중에 트리거가 돌면 반쪽 파일을 읽는다. 임시 이름으로 다 쓴 뒤 옮기게 한다.

셋째, 재전송이 중복을 만든다. 파일 원본은 경로로 새 파일을 판별하므로, 이름을 바꿔 다시 보낸 묶음은 그대로 들어온다. 고유 ID 로 중복을 제거하고, 상태가 끝없이 자라지 않도록 워터마크를 함께 건다.

실무에서 진짜 중요한 것

스트리밍은 작은 배치의 연속이다. 쿼리는 배치와 똑같이 쓴다.
offsets 는 처리 전에, commits 는 처리 뒤에 쓰인다. 그 사이에 죽으면 같은 범위를 다시 돈다.
파일 싱크의 결과는 _spark_metadata 에 적힌 파일만이다.
파일 원본은 경로로 새 파일을 판별한다. 내용 중복은 중복 제거로 막는다.
워터마크는 상태를 치우는 기준이다. 임계값 안쪽은 반드시 받고, 바깥은 대개 버린다.

다음 실습에서 할 것

착륙 폴더에 장치 이벤트 묶음을 넣고 availableNow 로 첫 배치를 돌려 체크포인트의 commits 와 출력 폴더의 _spark_metadata 에 0번 기록이 생기는 것을 확인한다. 새 묶음을 넣으면 그것만 처리되는 것, 두 묶음을 한꺼번에 넣으면 한 배치로 처리되는 것을 본다. 이름만 바꿔 재전송된 묶음이 중복으로 들어온 event_id 를 센 뒤, 새 체크포인트로 dropDuplicates 스트림을 돌려 고유 건수만 남긴다. 마지막으로 수정 시각을 지켜 복사한 두 번째 착륙 폴더에서 트리거당 파일 하나씩 10분 워터마크와 10분 창 집계를 돌려, 진행 기록으로 워터마크가 움직이는 모습과 40분쯤 늦게 온 이벤트가 버려지는 것을 본다.