📒 Spark(14)

📌 Spark Streaming

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

• Kafka, Kinesis, Flume, TCP 소켓 등의 다양한 소스에서 발생하는 데이터를 처리할 수있다.

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

Spark Streaming 동작 방식

• 데이터를 Micro Batch로 처리
• 계속해서 위의 과정을 반복(루프)
• 이렇게 읽은 데이터를 앞서 읽은 데이터에 Merge
• Batch마다 데이터 위치 관리(시작과 끝)
• Fault Tolerance와 데이터 재처리 관리(실패할 경우)

Spark Streaming 내부 동작

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

• DStream과 Structured Streaming 두 종류가 존재한다.

Spark Structured Streaming

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

Dstream	Structured Streaming
RDD 기반 스트리밍 처리	DataFrame 기반 스트리밍 처리
Spark SQL 엔진의 최적화 기능 사용불가능	Catalyst 기반 최적화 혜택을 가져간다.
이벤트 발생 시간 기반 처리 불가능	이벤트 발생 시간 기반을 처리 가능ㅎ다.
개발이 중단된 상태(RDD 기반 모두에 해당)	계속해서 기능이 추가되고 있다.

Source & Sink

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

Source

• Kafka, Amazon Kinesis, Apache Flume, TCP/IP 소켓, HDFS, File등을 Spark Structured Streaming에서 처리할 수 있도록 해준다.
  • 결국 Spark DataFrame으로 변환해준다.
  • ex) Kafka에서 Spark Structured Streaming으로데이터를 수집하려는 경우, Kafka Source를 사용하여 Kafka 클러스터에서 하나 이상의 토픽에서 데이터를 가져와 DataFrame으로 변환 가능하다.
• Spark Data Frame과 비교하면 readStream을 사용하는 점이 다르다

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

Sink

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

• 변환되거나 집계된 데이턱 어떻게 쓰이거나 소비되는지를 정의한다.

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

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

  • Append
  • Update
  • Complete : Full Refresh 느낌

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

Micro Batch Trigger Option

• Unspecified : default 모드, 현재 Micro Batch가 끝나면 다음 Batch가 바로 시작한다.

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

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

• Continuous : 새로운 저지연 연속 처리 모드에서 실행

---

📌 Spark Streaming 개발 환경 설정

Local Standalone Spark

• Spark Cluster Manager로 local[n] 지정

  • master를 local[n]으로 지정한다.
  • master는 클러스터 매니저를 지정하는데 사용한다.

• 주로 개발이나 간단한 테스트 용도이다.

• 하나의 JVM에서 모든 프로세스를 실행한다.

  • 하나의 Driver와 하나의 Executor가 실행된다.
  • 1+ Thread가 Executor안에서 실행된다.

• Executor안에 생성되는 Thread 수

  • local : 하나의 Thread만 생성
  • local[*] : 컴퓨터 CPU 수만큼 Thread 생성

Local Standalone Spark 설치

• Mac이라면 Mac Catalina 혹은 이후 버전기준

  • z Shell이 기본으로 사용된다 (그전에는 Bash shell)

• Java

  • JDK8/11 필요 : 터미널에서 java-version 명령으로 체크해야함
  • JAVA_HOME 환경변수를 Z Shell 시작 스크립트(~/.zshrc)에 등록한다.
    • echo export "JAVA_HOME=\$(//usr/libexec/java_home)" >> ~/.zshrc

• Spark 다운로드