Spark 구조적 스트리밍 설명
Spark 구조적 스트리밍은 인 메모리 프로세싱을 위한 인기 플랫폼이며, 일괄 처리 및 스트리밍을 위한 통합 패러다임을 가지고 있습니다. 사용자가 일괄 처리와 관련해 학습하고 사용하는 내용은 스트리밍을 위해서도 사용할 수 있어 데이터 일괄 처리부터 스트리밍까지 쉽게 확장할 수 있습니다. Spark Streaming은 단순히 Apache Spark 위에서 실행되는 엔진입니다.
구조적 스트리밍은 입력 데이터에 작업(예: 선택, 프로젝션, 집계, 창 작업 및 스트리밍 데이터 프레임과 참조 데이터 프레임의 조인)을 적용하는 장기 실행 쿼리를 만듭니다. 그런 다음, 사용자 지정 코드(예: SQL Database 또는 Power BI)를 사용하여 결과를 File Storage(Azure Storage Blob 또는 Data Lake Storage) 또는 데이터 저장소에 출력합니다. 또한 구조적 스트리밍은 로컬로 디버그하기 위해 콘솔에 출력을 제공하고, HDInsight에서 디버그하기 위해 생성된 데이터를 볼 수 있도록 메모리 내 테이블에 출력을 제공합니다.
테이블로서의 스트림
Spark 구조적 스트리밍은 데이터 스트림을 한정되지 않은 테이블로 상세하게 보여줍니다. 즉, 새 데이터가 들어옴에 따라 테이블은 계속 증가합니다. 이 입력 테이블은 장기 실행 쿼리를 통해 지속적으로 처리되며, 결과는 출력 테이블로 보내집니다.
구조적 스트리밍에서 데이터는 시스템에 들어오는 즉시 입력 테이블로 수집됩니다. 이 입력 테이블에 대해 작업을 수행하는 쿼리를 작성합니다(데이터 프레임 및 데이터 세트 API 사용). 쿼리 결과는 결과 테이블이라는 또 다른 테이블을 생성합니다. 결과 테이블에는 쿼리의 결과가 포함되며, 이 테이블에서 관계형 데이터베이스와 같은 외부 데이터 저장소에 대한 데이터를 가져옵니다. 데이터가 입력 테이블에서 처리되는 타이밍은 트리거 간격으로 제어됩니다. 트리거 간격은 기본적으로 0이며, 이 경우 데이터가 들어오는 즉시 구조적 스트리밍에서 처리하려고 시도합니다. 실제로 구조적 스트리밍은 이전 쿼리의 실행을 처리하는 즉시 새로 받은 데이터에 대해 또 다른 처리를 시작합니다. 트리거가 일정한 간격으로 실행되도록 구성하여 스트리밍 데이터를 시간 기반 일괄 처리로 처리할 수 있습니다.
결과 테이블의 데이터는 쿼리가 마지막으로 처리된 이후의 새 데이터만 포함하거나(추가 모드), 새 데이터가 생길 때마다 테이블을 새로 고쳐 스트리밍 쿼리가 시작된 이후의 모든 출력 데이터가 테이블에 포함되도록 할 수 있습니다(전체 모드).
추가 모드
추가 모드에서는 마지막 쿼리가 실행된 이후 결과 테이블에 추가된 행만 결과 테이블에 나타나고 외부 스토리지에 기록됩니다. 예를 들어 가장 간단한 쿼리는 단순히 입력 테이블의 모든 데이터를 변경되지 않은 상태로 결과 테이블에 복사합니다. 트리거 간격이 경과할 때마다 새 데이터가 처리되고 해당 새 데이터를 나타내는 행이 결과 테이블에 표시됩니다.
주가 데이터를 처리하는 시나리오를 살펴보겠습니다. 첫 번째 트리거가 00:01에 MSFT 주식에 대해 95달러의 가치로 이벤트 하나를 처리했습니다. 쿼리의 첫 번째 트리거에서는 00:01이 있는 행만 결과 테이블에 표시됩니다. 00:02에 다른 이벤트가 발생하는 경우 유일한 새 행은 00:02이 있는 행이므로 결과 테이블에는 이 하나의 행만 포함됩니다.
추가 모드를 사용하는 경우 쿼리에서 프로젝션(관심 있는 열 선택), 필터링(특정 조건과 일치하는 행만 선택) 또는 조인(정적 조회 테이블의 데이터로 데이터 보강)을 적용합니다. 추가 모드를 사용하면 관련된 새 데이터 요소만 외부 스토리지로 쉽게 푸시할 수 있습니다.
전체 모드
이번에는 전체 모드를 사용하여 동일한 시나리오를 살펴보겠습니다. 전체 모드에서는 모든 트리거에서 전체 출력 테이블을 새로 고쳐 가장 최근의 트리거 실행뿐만 아니라 모든 실행의 데이터가 테이블에 포함됩니다. 전체 모드를 사용하여 변경되지 않은 상태의 데이터를 입력 테이블에서 결과 테이블에 복사할 수 있습니다. 트리거된 모든 실행에서 새 결과 행이 모든 이전 행과 함께 표시됩니다. 출력 결과 테이블에는 쿼리가 시작된 이후에 수집된 모든 데이터가 저장되고 결국 메모리가 부족해집니다. 전체 모드는 들어오는 데이터를 어떤 방식으로든 요약하는 집계 쿼리에 사용하기 위한 것이므로 트리거가 있을 때마다 결과 테이블이 새 요약으로 업데이트됩니다.
지금까지 5초 분량의 데이터가 이미 처리되었으며 여섯 번째 초의 데이터를 처리한다고 가정합니다. 입력 테이블에는 00:01 및 00:03에 대한 이벤트가 있습니다. 이 예제 쿼리의 목표는 5초마다 주식의 평균 주가를 제공하는 것입니다. 이 쿼리를 구현하면 매 5초의 시간 범위에 속하는 모든 값을 사용하여 주가 평균을 계산하고 해당 간격에 대한 평균 주가 행을 생성하는 집계가 적용됩니다. 첫 번째 5초 시간 범위의 끝에 두 개의 튜플, 즉 (00:01, 1, 95) 및 (00:03, 1, 98)이 있습니다. 따라서 00:00-00:05 시간 범위의 집계는 평균 주가가 $96.50인 튜플을 생성합니다. 다음 5초 시간 범위에는 00:06 시간의 데이터 요소 하나만 있으므로 결과 주가는 $98입니다. 00:10 시간에서 전체 모드를 사용하는 경우 쿼리에서 새 행뿐만 아니라 집계된 모든 행을 출력하기 때문에 결과 테이블에는 00:00-00:05 및 00:05-00:10의 두 시간 범위에 대한 행이 있습니다. 따라서 새 시간 범위가 추가됨에 따라 결과 테이블은 계속 증가합니다.
전체 모드를 사용하는 모든 쿼리에서 테이블이 무제한으로 증가하지는 않습니다. 앞의 예제에서 범위당 주가 평균을 계산한 것이 아니라 주식별로 평균을 계산했다는 점을 생각해야 합니다. 결과 테이블에는 해당 디바이스로부터 받은 모든 데이터 요소에 걸쳐 평균 주가를 나타내는 고정된 수의 행(주식당 하나씩)이 있습니다. 새 주가가 수신되면 테이블의 평균이 항상 최신이 되도록 결과 테이블이 업데이트됩니다.
Spark 구조적 스트리밍의 이점은 무엇인가요?
금융 부문에서 거래 타이밍은 매우 중요합니다. 예를 들어 주식 거래 시 주식 시장에서 주식 거래가 발생하는 시점이나 거래를 수신하는 시점 또는 데이터를 읽는 시점 간 차이를 모두 고려해야 합니다. 금융 기관은 이러한 중요 데이터 및 이와 관련한 타이밍에 따라 움직입니다.
이벤트 시간, 지연 데이터 및 워터마크 처리
Spark 구조적 스트리밍은 이벤트 시간과 시스템에서 이벤트를 처리한 시간 간의 차이를 인식합니다. 각 이벤트는 테이블의 행이며 이벤트 시간은 행의 열 값입니다. 이렇게 하면 범위 기반 집계(예: 매분 이벤트 수)가 이벤트-시간 열에서 그룹화 또는 집계됩니다. 각 시간 범위는 그룹이고 각 행은 여러 범위/그룹에 속할 수 있습니다. 따라서 이러한 이벤트 시간 범위 기반 집계 쿼리는 데이터 스트림뿐만 아니라 정적 데이터 세트 모두에 일관되게 정의될 수 있으므로 데이터 엔지니어의 작업을 훨씬 수월하게 만듭니다.
또한 이 모델은 이벤트-시간을 기반으로 예상보다 늦게 도착한 데이터를 자연스럽게 처리합니다. Spark는 지연 데이터가 있는 경우 이전 집계를 업데이트하는 작업과 중간 상태 데이터 크기를 제한하기 위해 이전 집계를 정리하는 작업을 완벽히 컨트롤합니다. 또한 Spark 2.1부터 Spark는 지연 데이터의 임계값을 지정할 수 있는 워터마크 처리를 지원하고, 이에 따라 엔진이 이전 상태를 정리하도록 합니다.
최근 데이터 또는 모든 데이터를 업로드할 수 있는 유연성
이전 단원에서 설명한 바와 같이 Spark 구조적 스트리밍을 사용하여 작업할 때 추가 모드 또는 전체 모드 사용을 선택할 수 있습니다. 그러면 결과 테이블에 최신 데이터 또는 모든 데이터만 포함됩니다.
마이크로 일괄 처리에서 연속 처리로의 이동 지원
Spark 쿼리의 트리거 유형을 변경하면 프레임워크에 대한 다른 변경 없이 마이크로 일괄 처리에서 연속 처리로 이동할 수 있습니다. 다음은 Spark에서 지원되는 다양한 종류의 트리거입니다.
- 지정되지 않음(기본값). 트리거를 명시적으로 설정하지 않으면 쿼리는 마이크로 일괄 처리에서 실행되고 연속 처리됩니다.
- 고정 간격 마이크로 일괄 처리. 사용자가 설정한 되풀이 간격으로 쿼리가 시작됩니다. 새 데이터를 받지 못한 경우 마이크로 일괄 처리 프로세스가 실행되지 않습니다.
- 일회성 마이크로 일괄 처리. 쿼리가 단일 마이크로 일괄 처리를 실행한 다음 중지됩니다. 이는 이전 마이크로 일괄 처리 이후의 모든 데이터를 처리해야 하는 경우에 유용하며 연속으로 실행하지 않아도 되는 작업의 비용을 절감할 수 있습니다.
- 고정된 검사점 간격으로 연속 처리. 이 쿼리는 최소 한 번 이상 내결함성을 보장하면서 낮은(~1ms) 엔드투엔드 대기 시간을 가능하게 하는 새로운 대기 시간 연속 처리 모드로 실행됩니다. 이는 정확히 한 번만 보장할 수 있지만 최대 100ms의 대기 시간만 달성하는 기본값과 유사합니다.
일괄 처리 및 스트리밍 작업 결합
일괄 처리에서 스트리밍 작업으로의 이동을 단순화할 뿐 아니라 일괄 처리 및 스트리밍 작업을 결합할 수도 있습니다. 이 기능은 실시간 정보를 처리하는 동안 장기 기록 데이터를 사용하여 향후 추세를 예측하려는 경우에 특히 유용합니다. 주식의 경우 연간 또는 분기별 수익 발표의 변경 사항을 예측하기 위해 현재 가격 외에도 지난 5년 동안의 주식 가격을 확인하는 것이 좋습니다.
이벤트-시간 범위
Spark 구조적 스트리밍에서는 1일 또는 1분 시간 범위 등 설정한 간격에 따른 최고 주가 및 최저 주가 데이터를 캡처할 수 있도록 지원하며, 중복되는 범위도 지원합니다.
장애 복구를 위한 검사점 설정
오류 또는 의도적인 종료의 경우 이전 쿼리의 이전 프로세스 및 상태를 복구하고 중단된 위치에서 계속할 수 있습니다. 이 작업은 검사점 설정 및 미리 쓰기 로그를 사용하여 수행됩니다. 검사점 위치를 사용하여 쿼리를 구성할 수 있으며, 이 쿼리는 모든 진행 상황 정보(즉, 각 트리거에서 처리되는 오프셋의 범위)와 실행 중인 집계를 검사점 위치에 저장합니다. 이 검사점 위치는 HDFS 호환 파일 시스템에 있는 경로여야 하며 쿼리를 시작할 때 DataStreamWriter에서 옵션으로 설정할 수 있습니다.