Spark Streaming 소개

 

 

5. Spark Streaming 소개와 Kafka 연동

Spark Structured Streaming에 대해 알아보고 Kafka Topic을 사용하는 방법을 배워보자

 

 

 

 

 

 

Contents

1. Spark Streaming 소개
2. Spark 환경 설정
3. Streaming WordCount 예제 프로그램
4. Kafka Stream 예제 프로그램
5. 강의 마무리

 

 

 

 

 

Spark Streaming 소개

 

Spark Streaming이 무엇인지 알아보자

 

 

 

 

 

 

Spark의 등장

 

버클리 대학의 AMPLab에서 아파치 오픈소스 프로젝트로 2013년 시작

● 나중에 Databricks라는 스타트업 창업

 

하둡의 뒤를 잇는 2세대 빅데이터 기술

● YARN등을 분산환경으로 사용
● Scala로 작성됨

 

빅데이터 처리 관련 *다양한* 기능 제공

 

 

 

 

 

 

 

 

 

Spark 3.0의 구성

 

Spark Core

 

Spark SQL

 

Spark ML

Spark MLlib

 

Spark Streaming

 

Spark GraphX

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

Spark Streaming이란?

 

실시간 데이터 스트림 처리를 위한 Spark API

 

Kafka, Kinesis, Flume, TCP 소켓 등의 다양한 소스에서 발생하는 데이터 처리 가능

 

Join, Map, Reduce, Window와 같은 고급 함수 사용 가능

spark은 내부적으로 모든 데이터를 데이터프레임화하기 때문에 데이터 프레임에 적용되는 모든 함수를 Spark Streaming 상에서도 사용 가능합니다.

 

 

 

 

 

 

 

 

 

 

 

 

 

Spark Streaming 동작방식

● 데이터를 마이크로 배치로 처리 (완전 실시간 보다는 마이크로 배치)

● 계속해서 위의 과정을 반복 (루프)

● 이렇게 읽은 데이터를 앞서 읽은 데이터에 머지
● 배치마다 데이터 위치 관리 (시작과 끝)
● Fault Tolerance와 데이터 재처리 관리 (실패시)

 

 

 

 

 

 

 

 

 

 

 

Spark Streaming의 내부 동작

● Spark Streaming은 실시간 입력 데이터 스트림을 배치로 나눈 다음
● Spark Engine에서 처리하여 최종 결과 스트림을 일괄적으로 생성
    ○ DStream과 Structured Streaming 두 종류가 존재

 

 

DStream은 RDD기반으로 과거 방식, Structured Streaming은 미래 방식으로 Data Frame을 기반으로 사용

 

 

 

 

 

 

 

 

Spark Structured Streaming이란?

● Spark Streaming은 실시간 입력 데이터 스트림을 배치로 나눈 다음
● Spark Engine에서 처리하여 최종 결과 스트림을 일괄적으로 생성

 

 

 

 

 

 

 

 

 

 

 

Source & Sink

● 소스와 싱크는 외부 시스템(소스)에서 스트리밍 데이터를 수집하고 처리된 데이터를 외부 시스템 (싱크)으로 출력하는 것을 용이하게 하는 구성 요소

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

Source

 

Source는 Kafka, Amazon Kinesis, Apache Flume, TCP/IP 소켓, HDFS, File 등을 Spark Structured Streaming에서 처리할 수 있도록 해줌

○ 결국 Spark DataFrame으로 변환해줌 (매우 유용)
○ 예) Kafka에서 Spark Structured Streaming으로 데이터를 수집하려는 경우, Kafka Source를 사용하여 Kafka 클러스터에서 하나 이상의 토픽에서 데이터를 가져와 DataFrame으로 변환 가능

 

Spark DataFrame과 비교하면 readStream을 대신 사용하는 점이 다름

 

lines_df = spark.readStream \
  .format("socket") \
  .option("host", "localhost") \
  .option("port", "9999") \
  .load()

 

 

 

 

 

 

 

 

 

Sink

 

Sink는 Spark Structured Streaming에서 처리된 데이터를 외부 시스템이나 스토리지로 출력 가능하게 해줌

 

Sink는 변환되거나 집계된 데이터가 어떻게 쓰이거나 소비되는지를 정의

○ Source와 마찬가지로, Sink는 Kafka, HDFS, Amazon S3, Apache Cassandra, JDBC 데이터베이스 등과 같은 다양한 대상에 대해 사용 가능
○ 예) Kafka Sink를 사용하여 Spark Structured Streaming에서 처리된 데이터를 Kafka Topic으로
쓰는 것이 가능

 

OutputMode: 현재 Micro Batch의 결과가 Sink에 어떻게 쓰일지 결정

○ Append
○ Update: UPSERT 같은 느낌
○ Complete: FULL REFRESH 같은 느낌

 

 

 

 

 

 

 

 

 

 

 

Sink - 샘플코드

word_count_query = counts_df.writeStream \
  .format("console") \
  .outputMode("complete") \
  .option("checkpointLocation", "chk-point-dir") \
  .start()

 

 

 

 

 

 

 

 

 

 

 

 

전체 구조

 

 

 

 

 

 

 

 

 

 

 

Micro Batch Trigger Option

크게 보면 5가지, 적게 보면 4가지 옵션이 존재합니다.

 

Unspecified: 디폴트 모드. 현재 Micro Batch가 끝나면 다음 Batch가 바로 시작

 

Time Interval: 고정된 시간마다 Micro Batch를 시작. 현재 Batch가 지정된 시간을 넘어서 끝나면 끝나자마자 다음 Batch가 시작됨. 읽을 데이터가 없는 경우 시작되지 않음

Airflow 주기와 비슷

 

One Time => Available-Now: 지금 있는 데이터를 모두 처리하고 중단

One Time은 한번 돌고 끝입니다. Available-Now는 조금의 개선을 거친 것으로 비슷하지만 데이터를 migration하고 전체적으로 한번 프로세싱 하고 끝낼 때 유용합니다.

 

Continuous: 새로운 저지연 연속 처리 모드에서 실행. 아직 베타/실험 버전

마이크로배치보다 조금 더 실시간성을 띄도록 합니다. 아직은 시험단계입니다.

 

 

 

 

 

 

 

 

 

 

 

 

 

Micro Batch Trigger Option 예제

 

# Default trigger (runs micro-batch as soon as it can)
df.writeStream \
  .format("console") \
  .start()

# ProcessingTime trigger with two-seconds micro-batch interval
df.writeStream \
  .format("console") \
  .trigger(processingTime='2 seconds') \
  .start()

# Available-now trigger
df.writeStream \
  .format("console") \
  .trigger(availableNow=True) \
  .start()

# Continuous trigger with one-second checkpointing interval
df.writeStream
  .format("console")
  .trigger(continuous='1 second')
  .start()