Apache Spark Structured Streaming에서 지연 시간이 1초 미만으로 단축됩니다.

Apache Spark Structured Streaming 은 선도적인 오픈 소스 스트림 처리 플랫폼입니다. 또한 Databricks Lakehouse Platform에서의 스트리밍 을 구동하는 핵심 기술이며 배치 및 스트림 처리를 위한 통합 API를 제공합니다. 스트리밍 채택이 빠르게 증가함에 따라 다양한 애플리케이션에서 실시간 의사 결정을 위해 이를 활용하고자 합니다. 이러한 애플리케이션 중 일부, 특히 운영 환경에 있는 애플리케이션은 더 낮은 지연 시간을 요구합니다. Spark의 설계는 더 낮은 비용으로 높은 처리량과 사용 편의성을 지원하지만, 초 미만의 지연 시간에는 최적화되지 않았습니다.

이 블로그에서는 Structured Streaming의 고유한 처리 지연 시간을 줄이기 위해 오프셋 관리 부문에서 이룬 개선 사항을 집중적으로 살펴보겠습니다. 이러한 개선 사항은 단순하고 상태 비저장(stateless)인 실시간 모니터링 및 알림과 같은 운영 사용 사례를 주로 대상으로 합니다.

이러한 개선 사항에 대한 광범위한 평가 결과, 처리량이 초당 10만, 50만, 1백만 이벤트일 때 지연 시간이 700-900ms에서 150-250ms 로 68~75%, 즉 최대 3배까지 개선된 것으로 나타났습니다. 이제 Structured Streaming은 250ms 미만의 지연 시간을 달성할 수 있어 대부분의 운영 워크로드에 대한 SLA 요구 사항을 충족합니다.

이 아티클은 독자가 Spark Structured Streaming에 대한 기본적인 이해가 있다고 가정합니다. 자세한 내용은 다음 문서를 참조하세요.

https://www.databricks.com/spark/getting-started-with-apache-spark/streaming
https://docs.databricks.com/structured-streaming/index.html
https://www.databricks.com/glossary/what-is-structured-streaming
https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html

동기

Apache Spark Structured Streaming은 Apache Spark SQL 엔진을 기반으로 구축된 분산 스트림 처리 엔진입니다. 개발자가 배치 쿼리와 동일한 방식으로 스트리밍 쿼리를 작성하여 데이터 스트림을 처리할 수 있게 해주는 API를 제공하므로 스트리밍 애플리케이션에 대해 추론하고 테스트하기가 더 쉽습니다. Maven download에 따르면 Structured Streaming은 오늘날 가장 널리 사용되는 오픈 소스 분산 스트리밍 엔진입니다. 이러한 인기의 주된 이유 중 하나는 성능, 즉 몇 초 미만의 종단 간 지연 시간으로 더 낮은 비용에 높은 처리량을 제공한다는 점입니다. Structured Streaming은 사용자에게 throughput, 비용, 지연 시간 간의 절충점을 조절할 수 있는 유연성을 제공합니다.

엔터프라이즈에서 스트리밍 채택이 빠르게 증가함에 따라 다양한 애플리케이션에서 스트리밍 데이터 아키텍처를 사용하고자 하는 요구가 커지고 있습니다. 많은 고객과의 대화에서 일관된 초 미만 지연 시간이 필요한 사용 사례를 접했습니다. 이러한 낮은 지연 시간 사용 사례는 운영 알림 및 실시간 모니터링과 같은 애플리케이션에서 발생하며, 이를 '운영 워크로드'라고도 합니다. 이러한 워크로드를 Structured Streaming에 적용하기 위해 2022년에 Project Lightspeed라는 이름으로 성능 개선 이니셔티브를 시작했습니다. 이 이니셔티브는 처리 지연 시간을 개선하는 데 사용할 수 있는 잠재적인 영역과 기술을 식별했습니다. 이 블로그에서는 진행률 추적을 위한 오프셋 관리라는 개선 영역 중 하나와, 이를 통해 운영 워크로드에서 어떻게 초 미만 지연 시간을 달성하는지 자세히 설명합니다.

운영 워크로드란 무엇인가요?

스트리밍 워크로드는 크게 분석 워크로드와 운영 워크로드로 분류할 수 있습니다. 그림 1은 분석 워크로드와 운영 워크로드를 모두 보여줍니다. 분석 워크로드는 일반적으로 데이터를 실시간으로 수집, 변환, 처리, 분석하고 그 결과를 AWS S3, Azure Data Lake Gen2, Google Cloud Storage와 같은 객체 스토리지에서 지원하는 Delta Lake에 씁니다. 이러한 결과는 다운스트림 데이터 웨어하우징 엔진 및 시각화 도구에서 사용됩니다.

그림 1. 분석 워크로드와 운영 워크로드

분석 워크로드의 몇 가지 예는 다음과 같습니다.

• 고객 행동 분석: 마케팅 회사는 스트리밍 분석을 사용하여 실시간으로 고객 행동을 분석할 수 있습니다. 클릭스트림 데이터, 소셜 미디어 피드 및 기타 정보 소스를 처리하여 시스템은 고객을 보다 효과적으로 타겟팅하는 데 사용할 수 있는 패턴과 선호도를 감지할 수 있습니다.
• 감성 분석: 회사는 소셜 미디어 계정의 스트리밍 데이터를 사용하여 실시간으로 고객 감성을 분석할 수 있습니다. 예를 들어, 회사는 회사의 제품이나 서비스에 대해 긍정적이거나 부정적인 감성을 표현하는 고객을 찾을 수 있습니다.
• IoT 분석: 스마트 시티는 스트리밍 분석을 사용하여 실시간으로 교통 흐름, 공기 질 및 기타 지표를 모니터링할 수 있습니다. 도시 전역에 내장된 센서의 데이터를 처리함으로써 시스템은 추세를 감지하고 교통 패턴이나 환경 정책에 대한 결정을 내릴 수 있습니다.

반면에 운영 워크로드는 실시간으로 데이터를 수집 및 처리하고 비즈니스 프로세스를 자동으로 trigger합니다. 이러한 워크로드의 몇 가지 예시는 다음과 같습니다.

• 사이버 보안: 기업은 네트워크의 스트리밍 데이터를 사용하여 보안 또는 성능 문제를 모니터링할 수 있습니다. 예를 들어, 회사는 트래픽 급증이나 네트워크에 대한 무단 액세스를 감지하여 보안 부서에 알림을 보낼 수 있습니다.
• 개인 식별 정보 유출: 회사는 마이크로서비스 로그를 모니터링하고 파싱하여 개인 식별 정보(PII)가 유출되는지 감지하고, 유출된 경우 마이크로서비스 소유자에게 이메일로 알릴 수 있습니다.
• 엘리베이터 디스패치: 회사는 엘리베이터의 스트리밍 데이터를 사용하여 엘리베이터 비상 버튼이 활성화되는 시점을 감지할 수 있습니다. 활성화되면 추가 엘리베이터 정보를 조회하여 데이터를 보강하고 보안 담당자에게 알림을 보낼 수 있습니다.
• 사전 예방적 유지보수: 발전기의 스트리밍 데이터를 사용하여 온도를 모니터링하고 특정 threshold를 초과하면 감독자에게 알립니다.

운영 스트리밍 파이프라인은 다음과 같은 특징을 공유합니다.

• 지연 시간 기대치는 보통 1초 미만입니다.
• 파이프라인은 메시지 버스에서 읽습니다.
• 파이프라인은 일반적으로 데이터 변환 또는 데이터 보강을 통해 간단한 계산을 수행합니다.
• 파이프라인은 비즈니스 프로세스와의 다운스트림 통합을 위해 Apache Kafka 또는 Apache Pulsar와 같은 메시지 버스나 Apache Cassandra 또는 Redis와 같은 빠른 키-값 저장소에 기록합니다.

이러한 사용 사례에 대해 Structured Streaming을 프로파일링한 결과, 마이크로 배치 진행 상황을 추적하기 위한 오프셋 관리에 상당한 시간이 소요된다는 것을 확인했습니다. 다음 섹션에서는 기존의 오프셋 관리를 검토하고 이후 섹션에서 어떻게 개선했는지 간략히 설명하겠습니다.

오프셋 관리란 무엇인가요?

데이터가 어느 지점까지 처리되었는지 진행 상황을 추적하기 위해 Spark Structured Streaming은 진행률 표시기로 사용되는 오프셋을 유지하고 관리하는 데 의존합니다. 일반적으로 오프셋은 소스 커넥터에 의해 구체적으로 정의되는데, 이는 시스템마다 데이터의 진행률이나 위치를 나타내는 방식이 다르기 때문입니다. 예를 들어, 오프셋의 구체적인 구현은 파일의 데이터가 얼마나 처리되었는지를 나타내는 파일의 줄 번호가 될 수 있습니다. 이러한 오프셋을 저장하고 마이크로 배치의 완료를 표시하는 데 영구 logs(그림 2 참조)가 사용됩니다.

구조적 스트리밍(Structured Streaming)에서는 데이터가 마이크로 배치 단위로 처리됩니다. 각 마이크로 배치에 대해 두 가지 오프셋 관리 운영이 수행됩니다. 모든 마이크로 배치(batch)의 시작과 끝에 한 번씩입니다.

• 모든 마이크로 배치가 시작될 때(실제 데이터 처리가 시작되기 전), 대상 시스템에서 읽을 수 있는 새로운 데이터를 기반으로 오프셋이 계산됩니다. 이 오프셋은 체크포인트 디렉터리의 "offsetLog"라는 내구성 있는 로그에 유지됩니다. 이 오프셋은 "이번" 마이크로 배치에서 처리될 데이터의 범위를 계산하는 데 사용됩니다.
• 모든 마이크로 배치가 끝날 때마다 "이" 마이크로 배치가 성공적으로 처리되었음을 나타내기 위해 "commitLog"라고 하는 영구 로그에 항목이 영속화됩니다.

아래 그림 3은 현재 발생하는 오프셋 관리 운영을 보여줍니다.

또 다른 오프셋 관리 운영은 모든 마이크로 배치가 끝날 때 수행됩니다. 이 작업은 offsetLog와 commitLog가 무한정으로 커지지 않도록 두 로그에서 오래되고 불필요한 항목을 삭제/잘라내는 정리 작업입니다.

이러한 오프셋 관리 운영은 크리티컬 패스에서 데이터의 실제 처리와 함께 인라인으로 수행됩니다. 이는 이러한 운영의 기간이 처리 지연 시간에 직접적인 영향을 미치며, 운영이 완료될 때까지는 어떠한 데이터 처리도 이루어질 수 없음을 의미합니다. 이는 클러스터 활용률에도 직접적인 영향을 미칩니다.

저희는 벤치마킹과 성능 프로파일링을 통해, 특히 운영 알림 및 실시간 모니터링 사용 사례에 자주 사용되는 상태 비저장 단일 상태 파이프라인에서 오프셋 관리 운영이 처리 시간의 대부분을 차지할 수 있다는 사실을 발견했습니다.

구조적 스트리밍의 성능 개선

비동기 진행률 추적

이 기능은 진행률 추적을 위해 오프셋을 지속하는 데 따른 지연 시간 오버헤드를 해결하기 위해 만들어졌습니다. 이 기능을 활성화하면 Structured Streaming 파이프라인이 마이크로 배치 내의 실제 데이터 처리와는 비동기적으로 그리고 병렬로 진행률을 체크포인트, 즉 offsetLog 및 commitLog를 업데이트할 수 있습니다. 즉, 실제 데이터 처리가 이러한 오프셋 관리 운영에 의해 차단되지 않으므로 애플리케이션의 지연 시간이 크게 개선됩니다. 아래 그림 5는 오프셋 관리를 위한 이 새로운 동작을 보여줍니다.

비동기식 업데이트 수행과 함께 사용자는 진행률이 체크포인트되는 빈도를 구성할 수 있습니다. 이는 오프셋 관리 운영이 처리 가능한 속도보다 더 빠르게 발생하는 시나리오에 유용합니다. 이러한 현상은 파이프라인에서 실제 데이터 처리 시간이 오프셋 관리 운영에 비해 훨씬 적을 때 발생합니다. 이러한 시나리오에서는 오프셋 관리 운영의 백로그가 계속 증가하게 됩니다. 늘어나는 이 백로그를 막으려면 데이터 처리를 차단하거나 속도를 늦춰야 하며, 이는 본질적으로 처리 동작을 오프셋 관리 운영이 데이터 처리와 함께 인라인으로 실행될 때와 동일한 상태로 되돌립니다. 적절한 기본값이 설정되므로 사용자는 일반적으로 체크포인트 빈도를 구성하거나 설정할 필요가 없습니다. 체크포인트 간격이 길어질수록 장애 복구 시간도 늘어난다는 점에 유의하는 것이 중요합니다. 장애 발생 시 파이프라인은 이전의 성공적인 체크포인트 이후의 모든 데이터를 재처리해야 합니다. 사용자는 일반 처리 중의 낮은 지연 시간과 장애 발생 시의 복구 시간 사이의 이러한 트레이드오프를 고려할 수 있습니다.

이 기능을 활성화하고 구성하기 위해 다음과 같은 구성이 도입되었습니다.

• asyncProgressTrackingEnabled - 비동기 진행률 추적 활성화 또는 비활성화
  • Default: false
• asyncProgressCheckpointingInterval - 오프셋과 완료 커밋을 커밋하는 간격입니다.
  • Default: 1분

다음 코드 샘플은 이 기능을 활성화하는 방법을 보여줍니다.

참고: 이 기능은 Trigger.once 또는 Trigger.availableNow 와 함께 작동하지 않습니다. 이러한 트리거는 파이프라인을 수동 또는 예약된 방식으로 실행합니다. 따라서 비동기 진행률 추적은 적용되지 않습니다. 앞서 언급된 트리거를 사용하여 쿼리를 제출하면 실패합니다.

적용 가능성 및 제한 사항

현재 버전에는 기능을 발전시켜 나가면서 변경될 수 있는 몇 가지 제한 사항이 있습니다.

• 현재 비동기 진행률 추적은 Kafka 싱크를 사용하는 상태 비저장 파이프라인에서만 지원됩니다.
• 장애 발생 시 배치의 오프셋 범위가 변경될 수 있으므로 이 비동기 진행률 추적에서는 종단 간 정확히 한 번(exactly once) 처리가 지원되지 않습니다. 그러나 Kafka 싱크와 같은 많은 싱크는 최소 한 번(at-least once) 보장만 지원하므로 이는 새로운 제한 사항이 아닐 수 있습니다.

비동기 로그 제거

이 기능은 마이크로 배치 내에서 인라인으로 수행되던 로그 정리의 지연 시간 오버헤드를 해결하기 위해 만들어졌습니다. 이 Logs 정리/제거 운영을 비동기적으로 만들고 백그라운드에서 수행함으로써 실제 데이터 처리에 이 운영이 발생시키는 지연 시간 오버헤드를 제거할 수 있습니다. 또한 이러한 제거 작업은 모든 마이크로 배치마다 수행할 필요가 없으며 더 완화된 일정에 따라 수행될 수 있습니다.

참고로 이 기능/개선 사항은 어떤 유형의 파이프라인이나 워크로드에서 이를 사용할 수 있는지에 대한 제한이 없으므로, 모든 Structured Streaming 파이프라인에 대해 이 기능이 백그라운드에서 default로 활성화됩니다.

벤치마크

비동기 진행률 추적 및 비동기 로그 제거의 성능을 이해하기 위해 몇 가지 벤치마크를 만들었습니다. 벤치마크의 목표는 개선된 오프셋 관리가 엔드투엔드 스트리밍 파이프라인에서 제공하는 성능 차이를 이해하는 것입니다. 벤치마크는 두 가지 범주로 나뉩니다.

• Rate 소스에서 Stat 싱크로 - 이 벤치마크에서는 기본적이고 상태 비저장이며 통계를 수집하는 소스 및 싱크를 사용했습니다. 이는 외부 종속성 없이 코어 엔진 성능의 차이를 확인하는 데 유용합니다.
• Kafka 소스에서 Kafka 싱크로 - 이 벤치마크에서는 Kafka 소스에서 Kafka 싱크로 데이터를 이동합니다. 이는 프로덕션 시나리오에서 어떤 차이가 있을지를 확인하는 실제 시나리오와 유사합니다.

이 두 벤치마크 모두에서 다양한 데이터 입력 속도(초당 10만, 50만, 1백만 이벤트)에서 종단 간 지연 시간(50번째 백분위수, 99번째 백분위수)을 측정했습니다.

벤치마크 방법론

주요 방법론은 특정 상수 처리량으로 소스에서 데이터를 생성하는 것이었습니다. 생성된 레코드에는 레코드가 생성된 시점에 대한 정보가 포함되어 있습니다. 싱크 측에서는 Apache DataSketches 라이브러리를 사용하여 싱크가 레코드를 처리하는 시간과 각 배치에서 레코드가 생성된 시간 간의 차이를 수집합니다. 이는 지연 시간을 계산하는 데 사용됩니다. 모든 실험에 동일한 수의 노드를 가진 동일한 클러스터를 사용했습니다.

참고: Kafka 벤치마크의 경우, 클러스터의 일부 노드를 Kafka 실행 및 Kafka에 공급할 데이터 생성용으로 따로 할당했습니다. 레코드의 지연 시간은 레코드가 Kafka(싱크에서)에 성공적으로 게시된 후에만 계산합니다.

속도 소스-Stat 싱크 벤치마크

이 벤치마크에서는 Databricks 런타임(11.3)을 사용하여 7개의 워커 노드(i3.2xlarge - 4코어, 61GiB 메모리)로 구성된 Spark 클러스터를 사용했습니다. 각 개선 사항의 기여도를 정량화하기 위해 다음 시나리오에 대한 종단 간 지연 시간을 측정했습니다.

• 현재 구조적 스트리밍(Structured Streaming) - 이는 앞서 언급한 개선 사항이 없는 기준 지연 시간입니다.
• 비동기 로그 제거 - 비동기 로그 제거만 적용한 후의 지연 시간을 측정합니다.
• 비동기 진행률 - 비동기 진행률 추적을 적용한 후의 지연 시간을 측정합니다.
• 비동기 진행 + 비동기 로그 제거 - 두 개선 사항을 모두 적용한 후의 지연 시간을 측정합니다.

이러한 실험 결과는 그림 6, 7, 8에 나와 있습니다. 보시다시피, 비동기 로그 제거는 레이턴시를 일관적으로 약 50% 줄여줍니다. 마찬가지로 비동기 진행률 추적만으로도 지연 시간이 약 65% 개선됩니다. 함께 결합하면 지연 시간이 85~86% 감소하고 100ms 미만으로 떨어집니다.

Kafka 소스-Kafka 싱크 벤치마크

Kafka 벤치마크를 위해 작업자 노드 5개(i3.2xlarge - 코어 4개, 61GiB 메모리)로 구성된 Spark 클러스터, Kafka를 실행하기 위한 별도의 노드 3개 클러스터, Kafka 소스에 추가된 데이터를 생성하기 위한 추가 노드 2개를 사용했습니다. 저희 Kafka 토픽에는 40개의 파티션과 3의 복제 계수가 있습니다.

데이터 생성기는 데이터를 Kafka 토픽에 게시하고, Structured Streaming 파이프라인은 데이터를 소비하여 다른 Kafka 토픽으로 다시 게시합니다. 성능 평가 결과는 그림 9, 10, 11에 나와 있습니다. 보시다시피 비동기 진행 및 비동기 로그 제거를 적용하면 다양한 throughput에서 지연 시간이 65~75% 또는 3~3.5배 감소합니다.

성능 결과 요약

새로운 비동기 진행률 추적 및 비동기 로그 제거를 통해 두 구성 모두 지연 시간을 최대 3배까지 줄이는 것을 볼 수 있습니다. 함께 작동하면 모든 처리량에서 지연 시간이 크게 줄어듭니다. 차트는 또한 절약되는 시간이 일반적으로 일정한 시간(각 구성당 200~250ms)이며, 함께 사용하면 전반적으로 약 500ms를 단축하여 배치 계획 및 쿼리 처리에 충분한 시간을 남길 수 있음을 보여줍니다.

가용성

이러한 성능 향상은 DBR 11.3부터 Databricks Lakehouse Platform에서 사용할 수 있습니다. 비동기 로그 제거는 DBR 11.3 및 이후 릴리스에서 기본적으로 활성화됩니다. 또한 이러한 개선 사항은 오픈 소스 Spark에 기여되었으며 Apache Spark 3.4부터 사용할 수 있습니다.

향후 작업

비동기식 진행률 추적 기능에서 지원하는 워크로드 및 싱크 유형에는 현재 몇 가지 제한 사항이 있습니다. 향후 이 기능으로 더 많은 유형의 워크로드를 지원하는 방안을 검토할 예정입니다.

이는 Project Lightspeed의 일환으로 저희가 Structured Streaming에 구축하고 있는 예측 가능한 짧은 지연 시간 기능의 시작에 불과합니다. 또한, 저희는 개선의 여지가 있는 부분을 더 많이 찾기 위해 Structured Streaming을 계속 벤치마킹하고 프로파일링할 것입니다. 계속 지켜봐 주세요!

6월 26일부터 29일까지 샌프란시스코에서 열리는 Data and AI Summit에 참여하여 Project Lightspeed와 Databricks Lakehouse Platform의 데이터 스트리밍에 대해 자세히 알아보세요.

(이 글은 AI의 도움을 받아 번역되었습니다. 원문이 궁금하시다면 여기를 클릭해 주세요)