체크포인트 상태를 재구축할 필요 없이, 사기 탐지부터 실시간 감시에 이르기까지 가장 까다로운 상태 저장 스트림의 크기를 최적화하세요.
작성자: 탕감 바이야푸리, 제이 팔라니아판, B. Micheal Okutubo , Zifei Feng
프로덕션 환경에서 상태 저장형 Apache Spark™ Structured Streaming 쿼리를 실행해 본 사람이라면 누구나 결국 똑같이 곤란한 상황에 직면하게 됩니다.
몇 달 전에 쿼리를 시작했다고 가정해 보겠습니다. 당시에는 데이터 양이 많지 않아 기본값인 200개의 셔플 파티션을 그대로 적용하고 넘어갔습니다. 파이프라인은 원활하게 작동했습니다. 그러다 비즈니스가 성장하고 트래픽이 3배로 늘어났으며, 상태 저장소가 급격히 커졌습니다. 갑자기 이 200개의 파티션은 더 이상 적절한 크기가 아니게 됩니다. 일부 파티션에 데이터가 편중되어 과부하가 걸리고, 클러스터는 힘에 부치며, 모든 마이크로배치가 평소보다 더 오래 걸립니다.
그래서 당연히 다음과 같은 조치를 취합니다. spark.sql.shuffle.partitions 설정을 높이고 쿼리를 재시작합니다. 하지만 아무것도 바뀌지 않습니다.
스트림을 처음 시작할 때 파티션 수가 체크포인트에 고정되었기 때문에, 쿼리는 새로 설정한 값을 자동으로 무시합니다. 이전에는 새로운 파티션 수를 적용하는 유일한 방법이 기존 체크포인트를 버리고 처음부터 다시 시작하는 것뿐이었습니다. 하지만 상태 저장형 쿼리에서 이는 지금까지 신중하게 유지해 온 모든 누적 상태를 잃게 된다는 것을 의미합니다. 수백만 개의 계정을 추적하는 이상 거래 탐지 모델이나 며칠 동안의 윈도우 데이터를 보유하는 세션화 작업의 경우, 프로덕션 장애 검토 회의에서 "처음부터 다시 시작"이라는 말을 꺼내고 싶은 사람은 아무도 없을 것입니다.
Databricks Runtime 18 이상에서 제공되는 온디맨드 상태 재파티셔닝(퍼블릭 프리뷰)은 이러한 장벽을 없애줍니다. 이제 체크포인트 상태를 그대로 유지하면서 상태 저장형 스트리밍 쿼리의 파티션 수를 조정할 수 있습니다.
이는 집계, 스트림-스트림 조인, 중복 제거, 세션화 또는 transformWithState 등 어떤 작업을 실행하든 모든 상태 저장형 스트리밍 쿼리에 적용되며, 이상 거래 탐지부터 실시간 모니터링에 이르는 모든 워크로드에 적용됩니다.
Coveo와 같은 얼리 어답터의 경우, 스트리밍 인프라를 온디맨드로 적절한 크기로 조정할 수 있는 기능 덕분에 즉각적으로 상당한 운영 비용을 절감할 수 있었습니다.
Coveo에서는 시간이 지남에 따라 데이터 볼륨이 크게 변동하는 대규모 상태 저장형 스트리밍 파이프라인을 운영하고 있습니다. Databricks와 상태 재파티셔닝(State Repartitioning) 기능 덕분에 관련 Amazon S3 API 비용을 40% 절감했습니다. 이전에는 확장을 결정할 때마다 과도하게 리소스를 프로비저닝하거나 새로운 체크포인트에서 다시 빌드해야 하는 트레이드오프가 발생하여 스토리지 API 비용이 컴퓨팅 비용과 거의 비슷해졌습니다. 이제는 기존 상태를 손상시키거나 비용이 많이 드는 체크포인트 마이그레이션을 유발하지 않고도 수요 변화에 따라 자유롭게 확장할 수 있습니다.” —Alexis Chicoine, Coveo 수석 소프트웨어 개발자
Coveo의 결과가 Structured Streaming에 있어 왜 의미 있는 도약인지 이해하려면, 먼저 왜 파티션 수가 고정될 수밖에 없었는지 살펴보아야 합니다.
상태 저장형 스트리밍 쿼리는 상태를 상태 저장소에 보관하며, 이 상태는 물리적으로 분할됩니다. 스트림의 각 키(사용자 ID, 계정 번호, 윈도우 등)는 특정 파티션으로 해싱되며, 각 파티션의 데이터는 체크포인트 내의 자체 독립된 RocksDB 인스턴스에 저장됩니다. 파티션 수는 디스크에 있는 전체 상태 저장소의 레이아웃을 정의합니다.
재시작 사이에 단순히 파티션 수만 변경하면 해싱이 더 이상 일치하지 않게 됩니다. 예를 들어 이전에는 특정 파티션(예: 파티션 47)에 있던 키가 이제 다른 파티션(파티션 12)으로 해싱될 수 있지만, 누적된 상태는 여전히 원래 파티션의 파일에 남아 있게 됩니다. 결과적으로 쿼리는 자 체 메모리를 추적할 수 없게 됩니다. 이러한 소리 없는 데이터 손상을 방지하기 위해, Structured Streaming은 체크포인트 생성 시 파티션 수를 고정하고 이후 spark.sql.shuffle.partitions에 대한 변경 사항을 무시했습니다.
안전하지만 유연하지 못했습니다. 이로 인해 치러야 했던 두 가지 대가는 다음과 같습니다.
온디맨드 상태 재파티셔닝은 이전 디자인에서 불가능했던 작업을 수행하되, 새로운 파티션 수에 맞게 상태를 물리적으로 재분배함으로써 안전하게 두 가지 문제를 모두 해결합니다.
요구 사항은 간단합니다.
이것이 사전 요구 사항의 전부 입니다. 기본 상태 저장소가 있는 DBR 18을 사용 중이라면 이미 필요한 모든 준비가 완료된 것입니다.
메커니즘은 간단하며, 모든 스트리밍 개발자가 이미 알고 있는 패턴인 '중지, 재구성, 재시작'을 재사용합니다.
spark.sql.shuffle.partitions 대신 전용 구성인 spark.sql.streaming.stateStore.partitions을 설정하고 쿼리를 재시작합니다.
핵심 세부 사항은 새로운 구성 자체에 있습니다. 상태 저장형 쿼리의 경우, spark.sql.streaming.stateStore.partitions이 spark.sql.shuffle.partitions보다 우선순위를 갖습니다. 이것이 바로 기존 방식과 달리 변경 사항이 "유지"되도록 만드는 요인입니다.
쿼리가 재시작될 때 즉시 정상 처리를 재개하지는 않습니다. 먼저, 아직 대기 중인 마지막 계획된 마이크로배치가 있다면 이를 완료합니다. 그런 다음 일회성 재파티셔닝 작업을 수행합니다. 즉, 새로운 파티션 수에 맞춰 상태 데이터를 물리적으로 재분배하고 키를 올바른 새 위치로 다시 해싱하여 누락되거나 잘못 배치되는 데이터가 없도록 합니다. 재분배가 완료되면 쿼리는 요청한 파티션 수를 사용하여 평소와 같이 처리를 재개합니다.
이 재파티셔닝 단계가 이 기능의 핵심입니다. 이는 단순히 "숫자를 변경한 것"과 "상태를 새로운 레이아웃으로 안전하게 이동한 것"의 차이입니다.
재파티셔닝은 실행 시간이 상태의 양에 비례하는 실제 작업이므로, 진행 상황을 모 니터링하고 싶을 것입니다. Structured Streaming은 표준 진행 상황 보고를 통해 이를 제공합니다.
다음 마이크로배치가 완료되면, StreamingQueryProgress 이벤트에 재파티셔닝 작업의 소요 시간이 포함됩니다. 이벤트의 durationMs 메트릭에서 재파티셔닝 소요 시간을 밀리초 단위로 보고하는 controlBatch.REPARTITION 필드를 확인하세요.
상태 크기가 클수록 재파티셔닝 시간이 더 오래 걸리지만, 대부분의 워크로드에서는 몇 초 밖에 걸리지 않을 것으로 예상됩니다. 따라서 대규모 작업의 경우 소요 시간을 파악하기 위해 이 메트릭을 캡처할 가치가 있습니다. 이러한 이벤트를 읽는 방법에 대한 자세한 내용은 Databricks에서 Structured Streaming 쿼리 모니터링을 참조하세요.
ID별 이벤트의 텀블링 윈도우(tumbling-window) 수를 계산하는 간단한 집계를 통해 이를 구체적으로 살펴보겠습니다. 기본값인 200개의 파티션으로 시작한 다음, 이 워크로드에 필요한 것보다 많다고 판단하여 100개로 축소해 보겠습니다.
먼저, 현재 기본 파티션 수로 실행 중인 쿼리입니다.
이제 이 스트림을 한동안 관찰한 결과 200개의 파티션은 과도하다는 결론을 내렸습니다. 필요하지 않은 병렬 처리를 위해 조정 오버헤드(coordination overhead) 비용을 지불하고 있는 셈입니다. 쿼리를 중지하고 새로운 파티션 수를 설정한 다음, 동일한 옵션과 동일한 체크포인트로 재시작합니다.
재시작된 쿼리가 실행되면, 아직 대기 중인 마지막 계획된 마이크로배치가 있는 경우 이를 마무리하고, 재파티셔닝을 실행하여 상태를 200개 파티션에서 100개로 재분배한 다음, 모든 윈도우와 누적 합계를 완전히 보존한 상태로 계산을 계속 진행합니다. 역방향으로도 동일한 절차가 적용됩니다. 더 무거운 부하에서 확장(scale up)하려면 더 큰 숫자를 설정하기만 하면 됩니다.
동일한 접근 방식이 Spark Declarative Pipelines(SDP)에도 적용됩니다. 전체 과정을 살펴보려면 문서의 SDP 예시를 참조하세요.
온디맨드 상태 재분할은 일상적인 작업이라기보다는 튜닝 및 확장 도구에 가깝습니다. 이는 다음과 같은 몇 가지 주요 상황에서 유용합니다.
각 변경 시 일회성 재분할 일시 중지와 함께 중지 및 재시작이 필요하므로, 이를 계획된 유지 관리 작업으로 취급하세요. 잠시 처리가 중단되어도 괜찮은 시간대에 크기 조정을 계획하고, controlBatch.REPARTITION을(를) 모니터링하여 소요 시간을 확인한 후 쿼리가 정상적인 리듬으로 돌아오도록 하세요.
수년 동안 상태 저장 스트리밍 쿼리의 파티션 수는 처음에 한 번 결정하면 다시는 변경할 수 없거나, 상태를 처음부터 다시 구축하기 위해 큰 비용을 치러야 하는 결정이었습니다. 온디맨드 상태 재분할은 이러한 제약을 제거합니다. 새로운 파티션 수에 맞춰 상태를 안전하게 재분배함으로써, 시작할 때만 내릴 수 있었던 결정을 워크로드가 요구할 때마다 언제든지 다시 검토할 수 있는 결정으로 바꿔줍니다.
그 결과는 장기 실행 스트림 운영자가 그토록 원했던 것, 즉 상태를 잃지 않고 중지, 구성 변경, 재시작만으로 확장 요구 사항에 따라 쿼리 크기를 자유롭게 적정화할 수 있는 자유입니다.
온디맨드 상태 재분할은 RocksDB 상태 저장소 공급자를 사용하는 Databricks Runtime 18 이상에서 사용할 수 있습니다. 전체 참조는 상태 저장 스트리밍 쿼리를 위한 온디맨드 상태 재분할을 참조하세요.
(이 글은 AI의 도움을 받아 번역되었습니다. 원문이 궁금하시다면 여기를 클릭해 주세요)
블로그를 구독하고 최신 게시물을 이메일로 받아보세요.