2.4.3 Spark Streaming
Last updated
Last updated
Spark Streaming API 를 이용해 사용자는 실시간 데이터 처리를 수행할 수 있습니다. 이번 챕터에서는 Spark Streaming 의 기본적인 개념과 사용법에 대해 알아봅니다.
아래는 Structured Streaming API 를 이용해 작성된 Word Count (단어 수 세기) 예제 입니다. 챕터들을 지나며 아래의 주제들을 하나씩 알아보겠습니다.
Spark 3.2.0 기준으로 현재 두 가지 종류의 API 를 지원합니다.
Spark Streaming (Low-Level API, DStream)
Spark Structured Streaming (High-Level API)
Spark Streaming (Low-Level API, DStream) 은 Low-level API 로서 다음과 같은 특징이 있습니다.
Micro Batch 모드로 동작합니다. 즉, 입력 데이터를 일정 시간동안 데이터를 모아 사용자가 처리할 수 있도록 RDD 로 제공합니다.
사용자는 RDD 로 들어오는 데이터를 RDD API 를 이용해 처리합니다.
아래는 DStream API 를 이용한 코드 샘플입니다.
Spark Structured Streaming 은 High-level API 로서 다음과 같은 특징이 있습니다.
RDD 를 직접 다루지 않고, Spark SQL 의 최적화 등 이점을 누리며 편리한 DataFrame / Dateset API 를 사용할 수 있습니다.
Micro Batch 이외에도 Continuous Processing 을 지원합니다 (Experimental)
아래는 Structured Streaming API 를 이용한 코드 샘플입니다.
만약 Streaming 이 아니라 Batch (1회 적재 후 종료) 를 패턴으로 사용하고 싶다면 writeStream 대신 Generic Load / Save API 와 동일한 write 및 save 를 사용할 수 있습니다. Spark 는 Batch 와 Streaming 의 Sink 로 Kafka 를 지원합니다.
Spark Structured Streaming 을 이용하면 Batch / Stream 데이터 적재를 동일한 가공 API (DataFrame, Dataset) 을 이용해서 수행할 수 있습니다. API 의 통합이 사용자에게 언제 유용할까요?
다음 경우를 가정해 봅시다.
(A) Kafka Topic 의 Message 를 읽어 가공한 후 다른 Storage 로 적재합니다 (S3, Dynamo, Kafka 등)
(B) Processing 과정에서 실패한 Kafka Topic 내 Message 는 추후 다시 처리하기 위해 Dead Letter Queue (DLQ, 또 다른 Kafka Topic) 으로 보냅니다.
(C) DLQ Kafka Topic 내 Message 는 Retention 에 의해 삭제될 수 있기 때문에, DLQ 에 쌓인 메세지는 S3 에 적재합니다.
시간이 지나 S3 에 쌓인 데이터를 (Batch Source) 읽어 문제를 해결한 후 (A) 에서 보내려고 했던 Storage 에 다시 보낼 수 있습니다. 이 때 두 가지 옵션이 있을 수 있습니다.
S3 데이터를 Spark Batch Application 에서 읽어 최종 Storage 인 (A) 로 직접 적재합니다.
만약 (A) 에서 사용되는 Spark Streaming 에서 메모리에 State 등을 활용한다면, S3 데이터를 읽어 (A) Kafka Topic 으로 데이터를 흘려 보내고 Spark Streaming 이 이 메세지를 다시 처리하게끔 할 수 있습니다.
즉, (2) 번 옵션은 Spark Batch Application 이 S3 데이터를 읽어 원본 Kafka Topic 으로 다시 보내고, 이것을 기존의 Spark Steraming Application 이 처리합니다.
단, 원본 Kafka Topic 이 다른 Consumer 들에 의해 공유되는 경우 (Inventory 등) 별도 작업 복구를 위한 Kafka Topic 을 만들고 Spark Streaming Application 이 이것을 사용할 수 있습니다.
Spark Streaming (DStream API) 또는 Structured Streaming 에서는 Micro-batch Processing 모드를 기본적으로 지원합니다. Structured Streaming 의 경우에는 Micro-batch Processing 모드 이외에도 Continuous Processing 을 실험적으로 (Experimental) 지원합니다. 두 가지 프로세싱 모드의 차이점은 다음과 같습니다.
Micro-batch Processing 모드의 경우 들어오는 데이터를 일정 기간 동안 모았다가 (Batch) 처리합니다.
Continuous Processing 모드의 경우 지연 없이 바로바로 처리합니다.
아래 그림을 통해 두 가지 Processing 모드의 차이점을 살펴볼 수 있습니다.
Micro-batch Processing 모드와 Continuous Processing 모드의 차이점은 무엇일까요?
개별 Record 의 Latency 관점에서 생각해보고
또 다른 Streaming Framework 인 Flink 는 어떤 Processing 모드를 지원하는지 찾아봅시다.
Streaming Application 은 Kafka 로 부터 데이터를 받아 다양한 곳에 적재할 수 있습니다. 위 그림과 같이 S3 에 데이터를 적재할 수도 있고, 다시 Kafka 로 보내 다른 실시간 처리를 위해 사용할 수도 있습니다.
이 챕터에서는 크게 결과 저장소의 종류를 기준으로 4 가지 사례로 나누어 알아봅니다.
S3 에 데이터를 적재합니다. 가장 기본적인 적재 방식으로서 Client / Server 의 이벤트 데이터나 CDC (Storage) 로 부터 온 데이터를 S3 에 내려 Query Engine 을 통해 조회하거나 추가적인 가공을 위해 보관합니다.
하나 이상의 Kafka Topic 에서 데이터를 읽어 (Join) 처리 후 다른 Kafka Topic 으로 전송합니다. 때로는 다른 Kafka Cluster 의 Topic 으로 데이터를 전송할 수도 있습니다. Druid 와 같은 스토리지에서 이 Topic 의 데이터를 읽어 SQL 형태로 사용할 수 있습니다. Druid 는 기본적인 가공은 가능하나 복잡한 Join 및 Unnesting 등은 수행하기 어려우므로 Streaming Application 에서 가공하고 이것을 결과 Topic 에 저장한 뒤 Druid 와 같은 Storage 에서 읽어 사용합니다.
Kafka Topic 데이터를 읽어 External Key-Value 저장소에 지속적으로 '상태' 를 업데이트 합니다. (e.g., 상품의 재고 상태 등) 이 경우 '상태' 는 Stream Application 외부에 존재합니다. 경우에 따라 Stream (Event) 과 Batch (Meta) 데이터를 Join 할 수 있습니다. Batch 데이터는 1일 1회 갱신하며, 실시간으로 인입되는 Stream 데이터와 Join 해 결과를 만듭니다.
Kafka Topic 데이터를 읽어 Stream Application 내에 In-memory '상태' (State) 를 만든뒤 이 상태 값을 바탕으로 연산을 수행하고 결과를 내보냅니다. Output 은 다른 Kafka Topic 일 수 있습니다. 단, In-memory State 는 '메모리' 에 저장하므로 사이즈 제한이 있고, Checkpoint 등에 저장되어 안전하게 보일지라도 외부에서 사용 가능한 데이터가 아니므로 단일 Streaming Application 내에서만 활용 가능해, 범용성이 크게 집니다.
일반적으로 Streaming 의 결과물이 다른 팀에 의해서 소비된다면 Latency 가 매우 중요하므로 (@torreswoo) Streaming Application 의 Output 은 Storage 가 아니라 Kafka Topic 인 경우가 많습니다.
Stream Processing 은 실시간 데이터를 처리할 수 있지만 매우 물리 / 노동 비용 관점에서 고비용입니다. 다음 측면을 고려해 봅시다.
지속적으로 Streaming Appilcation 이 떠있어야 합니다. 특성상 Streaming Application 은 트래픽이 튀는 상황에서도 '안전성' 을 보장해야 하므로 일정량 이상의 여분 리소스가 필요합니다. (경우에 따라 Auto Scaling 을 사용할 수도 있습니다.)
지속적으로 Streaming Application 이 떠 있다는 말은 새벽에 문제가 생길 경우에도 담당자가 대응을 해야한다는 의미입니다. 자동 복구 등의 프로세스를 갖출 수 있겠지만, 예외적인 경우에는 사람의 간섭이 필요한 복구가 존재할 수 있습니다. (정책적 결정이 필요한 경우 등) Streaming Application 의 숫자가 늘어날수록 노동 인력에게 가해지는 부하가 늘어납니다.
Streaming Application 을 위한 기반 데이터 인프라가 필요합니다. Kafka / Zookeeper 등이 필요하고 Spark / Flink Streaming 을 위해 Cluster Manager 가 필요합니다.
단일 Streaming Application 을 위해 복구용 Topic 과 Batch Application, Dead Letter 큐를 S3 에 내리기 위한 보존용 Application 등 여러 Application 필요할 수 있습니다.
따라서 Streaming Application 은 일반적으로 '실시간 데이터' 를 사용해 효과가 혹은 혜택이 있는 경우에만 사용하는게 바람직합니다.
이제 각 패턴을 하나씩 살펴봅시다.
Kafka Topic 에서 데이터를 읽어 어디론가 저장하는 단순한 패턴입니다.
Kafka Topic 을 Spark 읽어 S3 (HDFS, GCS) 로 저장할 수 있습니다.
Kafka Topic 을 다른 Kafka Broker 로 Relay 하거나 Kinesis To Kafka Relay 일 수 있습니다.
Client Event 에 Session ID 발급 / Event 전처리 등 추가적인 가공이 들어갈 수 있기 때문에 Kafka Connect 처럼 단순 Relay 를 제공하는 프레임워크보다는 가공 로직을 사내 기준에 맞추어 자유롭게 변경할 수 있는 Spark / Flink 프레임워크가 선호됩니다.
Kafka Topic 을 용도에 맞추어 가공해 사용하는 패턴입니다.
여러 Topic 을 Union / Join 할 수 있습니다. 예를 들어 Client Web / Client App 이벤트의 포맷이 서로 다르며, 각기 다른 Kafka Topic 으로 전송될 때 두개의 Kafka Stream 을 공통 포맷으로 변환한 뒤 Spark DataFrame 에서 Join 할 수 있습니다.
이렇게 가공된 통합 Client Event 는 Kafka 로 보내져 Druid 등에서 소비되거나 직접 Storage 로 적재될 수 있습니다
또 다른 가공 패턴으로는 메타 테이블 + 이벤트 스트림을 Join 하는 케이스입니다. (Stream-Static Join)
상품의 메타를 Batch 를 통해 읽어오고
상품의 사용자 이벤트를 Stream 을 통해 읽어 Join 해 사용하는 방식입니다.
다만 상품의 메타 데이터는 일별로 갱신될 수 있으므로 매번 즉시 쿼리하기 보다는 캐싱 후 어떻게 읽어온 배치 데이터를 갱신할것인지 주의해야 합니다.
Stream-Stream Join 을 이용하면 다음과 실시간 Attribution 집계와 같은 작업을 해낼 수 있습니다.
Kafka Topic 에서 데이터를 Streaming Application 읽어 원하는 형태로 가공 후 원하는 Write / Read Capacity 와 Latency 를 제공하는 Storage 를 골라 데이터를 적재할 수 있습니다.
(User Profile) 사용자 마다의 최근 이벤트 N 개 (검색, 상품 뷰, 주문 등) 를 시간순으로 정렬해 State 로 만들어 최신의 State 를 Dynamo 등에 Write 할 수 있습니다.
(Product Profile) 상품의 재고 변동이나 상품 기준의 집계 메트릭 (주문, 조회 수, 범위 기간당 노출 대비 클릭률 등) 을 계산해 Storage 에 적재할 수 있습니다.
사용하는 Storage 의 Latency 및 Transaction, Upsert 등 구문 지원 여부에 따라 Streaming State 를 사용하지 않고 외부 저장소를 State 로 쓸 수 있습니다. (Get 후 Upsert)
아래 그림은 mapGroupsWithState 또는 flatMapGroupsWithState 를 사용해 Structured Streaming 에서 State 를 만드는 것을 도식화 한 그림입니다.
Spark Application 은 분산 처리를 수행하며 Executor 의 경우 N 개가 될 수 있으므로, 이 N 개 Executor 의 Memory 를 활용해 State 를 만들어 계산에 활용할 수 있습니다.
즉 Spark Executor 내 Memory 를 State 저장소로 활용해 외부 저장소 조회 없이 빠른 집계를 수행할 수 있습니다.
빠른 집계를 위해 In-memory State 를 쓰는 만큼 주된 Output 은 Kafka Topic 입니다.
다만 Memory 사이즈에 제한이 있으므로 Key 는 TTL 등을 통해 무한히 늘어나지 않도록 지정해야 합니다.
이번 챕터에서 나온 개념 또는 용어 대한 정리입니다.
Micro-batch Processing Mode
Continuous Processing Mode
Dead Letter Queue
Flink (Streaming Framework)
AWS Kinesis
Key Value Storage (Dynamo, HBase, ...)
