스트리밍 (Structured Streaming)

스트리밍 (Structured Streaming)

참고: 이 문서는 Spark의 스트리밍 엔진에 대한 가이드입니다. Spark Streaming은 이전 세대의 스트리밍 엔진으로, 더 이상 업데이트되지 않는 레거시(legacy) 프로젝트예요. 현재 Spark에는 더 새롭고 쓰기 쉬운 Structured Streaming이라는 스트리밍 엔진이 있죠. 스트리밍 애플리케이션이나 파이프라인을 만들 때는 Spark Structured Streaming을 사용하는 것을 권장합니다. 자세한 내용은 Structured Streaming Programming Guide를 참고하세요.

개요 (Overview)

Spark Streaming은 핵심 Spark API의 확장으로, 실시간 데이터 스트림에 대해 확장 가능하고 높은 처리량이며 결함에 강한 스트림 처리를 가능하게 해줘요. Kafka, Kinesis, TCP 소켓 같은 다양한 소스에서 데이터를 받아들이고, map, reduce, join, window 같은 고수준 함수로 표현한 복잡한 알고리즘으로 처리할 수 있습니다. 처리된 데이터는 파일시스템, 데이터베이스, 실시간 대시보드로 내보낼 수 있어요. 게다가 Spark의 머신러닝과 그래프 처리 알고리즘도 데이터 스트림에 적용할 수 있죠.

내부적으로는 이렇게 동작해요. Spark Streaming은 실시간 입력 데이터 스트림을 받아서 배치(batch) 단위로 나누고, 이 배치들을 Spark 엔진이 처리해서 최종 결과 스트림을 배치 단위로 생성해요.

Spark Streaming이 제공하는 핵심 추상화는 **discretized stream(이산화 스트림, 줄여서 DStream)이에요. DStream은 연속적인 데이터 스트림을 나타내죠. DStream은 Kafka, Kinesis 같은 소스에서 온 입력 데이터 스트림으로 만들 수도 있고, 다른 DStream에 고수준 연산을 적용해서 만들 수도 있습니다. 내부적으로 DStream은 RDD의 연속된 시퀀스로 표현돼요.

이 가이드는 DStream으로 Spark Streaming 프로그램을 어떻게 작성하는지 보여줍니다. Scala, Java, 그리고 Python(Spark 1.2부터 도입)으로 작성할 수 있고, 모두 이 가이드에서 다루고 있어요. 가이드 곳곳에 언어별 코드 조각을 선택할 수 있는 탭이 있으니 참고하세요.

참고: Python에는 없거나 다르게 동작하는 API가 몇 가지 있어요. 가이드 곳곳에서 Python API 태그로 이 차이점을 표시해 두었습니다.

빠른 예제 (A Quick Example)

자세한 내용으로 들어가기 전에, 간단한 Spark Streaming 프로그램이 어떤 모습인지 먼저 살펴볼게요. TCP 소켓에서 데이터를 받아주는 서버가 보내는 텍스트 데이터에서 단어 수를 세는 프로그램을 만든다고 해볼게요. 이렇게 하면 됩니다.

먼저 모든 스트리밍 기능의 진입점인 StreamingContext를 임포트해요. 두 개의 실행 스레드와 1초 배치 간격을 가진 로컬 StreamingContext를 만듭니다.

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

# Create a local StreamingContext with two working thread and batch interval of 1 second
sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, 1)

이 컨텍스트를 사용해서, 호스트명(예: localhost)과 포트(예: 9999)로 지정한 TCP 소스에서 오는 스트리밍 데이터를 나타내는 DStream을 만들 수 있어요.

# Create a DStream that will connect to hostname:port, like localhost:9999
lines = ssc.socketTextStream("localhost", 9999)

이 lines DStream은 데이터 서버에서 받는 데이터의 스트림을 나타내요. 이 DStream의 각 레코드는 텍스트 한 줄입니다. 이제 줄을 공백 기준으로 나눠 단어로 만들게요.

# Split each line into words
words = lines.flatMap(lambda line: line.split(" "))

flatMap은 일대다(one-to-many) DStream 연산이에요. 소스 DStream의 각 레코드에서 새 레코드를 여러 개 생성해서 새 DStream을 만듭니다. 여기서는 각 줄이 여러 단어로 나뉘고, 단어들의 스트림이 words DStream으로 표현돼요. 이제 이 단어들을 세어볼게요.

# Count each word in each batch
pairs = words.map(lambda word: (word, 1))
wordCounts = pairs.reduceByKey(lambda x, y: x + y)

# Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.pprint()

words DStream은 다시 (word, 1) 쌍의 DStream으로 매핑(일대일 변환)되고, 이어서 리듀스되어 각 배치 안에서 단어의 빈도를 얻어요. 마지막으로 wordCounts.pprint()가 매초 생성된 몇 개의 카운트를 출력합니다.

이 줄들을 실행했을 때 잊지 말아야 할 점이 있어요. 이 시점에서 Spark Streaming은 시작된 뒤 수행할 계산을 세팅만 하고, 실제 처리는 아직 시작되지 않아요. 모든 변환을 세팅한 다음 처리를 시작하려면 마지막으로 이렇게 호출해야 해요.

ssc.start()             # Start the computation
ssc.awaitTermination()  # Wait for the computation to terminate

전체 코드는 Spark Streaming 예제 NetworkWordCount에서 찾을 수 있어요.

먼저 Spark Streaming 클래스의 이름들과, 다른 클래스(예: DStream)에 유용한 메서드를 추가해주는 StreamingContext의 일부 암시적 변환을 환경으로 임포트해요. StreamingContext는 모든 스트리밍 기능의 메인 진입점입니다. 두 개의 실행 스레드와 1초 배치 간격을 가진 로컬 StreamingContext를 만듭니다.

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3

// Create a local StreamingContext with two working thread and batch interval of 1 second.
// The master requires 2 cores to prevent a starvation scenario.

val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
val ssc = new StreamingContext(conf, Seconds(1))

이 컨텍스트를 사용해서, 호스트명(예: localhost)과 포트(예: 9999)로 지정한 TCP 소스에서 오는 스트리밍 데이터를 나타내는 DStream을 만들 수 있어요.

// Create a DStream that will connect to hostname:port, like localhost:9999
val lines = ssc.socketTextStream("localhost", 9999)

이 lines DStream은 데이터 서버에서 받는 데이터의 스트림을 나타내요. 이 DStream의 각 레코드는 텍스트 한 줄입니다. 이제 줄을 공백 문자 기준으로 나눠 단어로 만들게요.

// Split each line into words
val words = lines.flatMap(_.split(" "))

flatMap은 일대다(one-to-many) DStream 연산이에요. 소스 DStream의 각 레코드에서 새 레코드를 여러 개 생성해서 새 DStream을 만듭니다. 여기서는 각 줄이 여러 단어로 나뉘고, 단어들의 스트림이 words DStream으로 표현돼요. 이제 이 단어들을 세어볼게요.

import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
// Count each word in each batch
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)

// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print()

words DStream은 다시 (word, 1) 쌍의 DStream으로 매핑(일대일 변환)되고, 이어서 리듀스되어 각 배치 안에서 단어의 빈도를 얻어요. 마지막으로 wordCounts.print()가 매초 생성된 몇 개의 카운트를 출력합니다.

이 줄들을 실행했을 때 잊지 말아야 할 점이 있어요. 이 시점에서 Spark Streaming은 시작된 뒤 수행할 계산을 세팅만 하고, 실제 처리는 아직 시작되지 않아요. 모든 변환을 세팅한 다음 처리를 시작하려면 마지막으로 이렇게 호출해야 해요.

ssc.start()             // Start the computation
ssc.awaitTermination()  // Wait for the computation to terminate

전체 코드는 Spark Streaming 예제 NetworkWordCount에서 찾을 수 있어요.

먼저 모든 스트리밍 기능의 메인 진입점인 JavaStreamingContext 객체를 만들어요. 두 개의 실행 스레드와 1초 배치 간격을 가진 로컬 StreamingContext를 만듭니다.

import org.apache.spark.*;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.*;
import org.apache.spark.streaming.api.java.*;
import scala.Tuple2;

// Create a local StreamingContext with two working thread and batch interval of 1 second
SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));

이 컨텍스트를 사용해서, 호스트명(예: localhost)과 포트(예: 9999)로 지정한 TCP 소스에서 오는 스트리밍 데이터를 나타내는 DStream을 만들 수 있어요.

// Create a DStream that will connect to hostname:port, like localhost:9999
JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);

이 lines DStream은 데이터 서버에서 받는 데이터의 스트림을 나타내요. 이 스트림의 각 레코드는 텍스트 한 줄입니다. 이제 줄을 공백 기준으로 나눠 단어를 만들게요.

// Split each line into words
JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());

flatMap은 소스 DStream의 각 레코드에서 새 레코드를 여러 개 생성해서 새 DStream을 만드는 DStream 연산이에요. 여기서는 각 줄이 여러 단어로 나뉘고, 단어들의 스트림이 words DStream으로 표현돼요. 참고로 변환을 FlatMapFunction 객체로 정의했죠. 앞으로 알아가겠지만, Java API에는 이런 식으로 DStream 변환을 정의하는 데 도움을 주는 편의 클래스가 여럿 있어요.

이제 이 단어들을 세어볼게요.

// Count each word in each batch
JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));
JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey((i1, i2) -> i1 + i2);

// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print();

words DStream은 PairFunction 객체를 사용해 (일대일 변환으로) (word, 1) 쌍의 DStream으로 매핑돼요. 그다음 Function2 객체를 사용해 리듀스되어 각 배치 안에서 단어의 빈도를 얻어요. 마지막으로 wordCounts.print()가 매초 생성된 몇 개의 카운트를 출력합니다.

이 줄들을 실행했을 때 잊지 말아야 할 점이 있어요. 이 시점에서 Spark Streaming은 시작된 뒤 수행할 계산을 세팅만 하고, 실제 처리는 아직 시작되지 않아요. 모든 변환을 세팅한 다음 처리를 시작하려면 마지막으로 start 메서드를 호출해야 해요.

jssc.start();              // Start the computation
jssc.awaitTermination();   // Wait for the computation to terminate

전체 코드는 Spark Streaming 예제 JavaNetworkWordCount에서 찾을 수 있어요.

이미 Spark를 다운로드하고 빌드했다면, 이 예제를 이렇게 실행할 수 있습니다. 먼저 대부분의 유닉스 계열 시스템에 있는 작은 유틸리티인 Netcat을 데이터 서버로 실행해야 해요.

$ nc -lk 9999

그다음 다른 터미널에서 예제를 시작하면 됩니다.

$ ./bin/spark-submit examples/src/main/python/streaming/network_wordcount.py localhost 9999

$ ./bin/run-example streaming.NetworkWordCount localhost 9999

$ ./bin/run-example streaming.JavaNetworkWordCount localhost 9999

그러면 netcat 서버를 실행 중인 터미널에 입력한 모든 줄이 매초마다 세어져서 화면에 출력돼요. 대략 이런 모습입니다.

# TERMINAL 1: RUNNING Netcat

$ nc -lk 9999

hello world

...

# TERMINAL 2: RUNNING network_wordcount.py

$ ./bin/spark-submit examples/src/main/python/streaming/network_wordcount.py localhost 9999
...
-------------------------------------------
Time: 2014-10-14 15:25:21
-------------------------------------------
(hello,1)
(world,1)
...

# TERMINAL 2: RUNNING NetworkWordCount

$ ./bin/run-example streaming.NetworkWordCount localhost 9999
...
-------------------------------------------
Time: 1357008430000 ms
-------------------------------------------
(hello,1)
(world,1)
...

# TERMINAL 2: RUNNING JavaNetworkWordCount

$ ./bin/run-example streaming.JavaNetworkWordCount localhost 9999
...
-------------------------------------------
Time: 1357008430000 ms
-------------------------------------------
(hello,1)
(world,1)
...

기본 개념 (Basic Concepts)

이제 간단한 예제를 넘어서서, Spark Streaming의 기초를 조금 더 자세히 살펴볼게요.

링킹 (Linking)

Spark처럼 Spark Streaming도 Maven Central에서 사용할 수 있어요. 나만의 Spark Streaming 프로그램을 작성하려면 SBT 또는 Maven 프로젝트에 다음 의존성을 추가해야 합니다.

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.13</artifactId>
    <version>4.2.0</version>
    <scope>provided</scope>
</dependency>

libraryDependencies += "org.apache.spark" % "spark-streaming_2.13" % "4.2.0" % "provided"

Kafka나 Kinesis처럼 Spark Streaming 핵심 API에는 없는 소스에서 데이터를 받아들이려면, 해당하는 spark-streaming-xyz_2.13 아티팩트를 의존성에 추가해야 해요. 예를 들어 흔히 쓰는 것들이 이렇습니다.

소스	아티팩트
Kafka	spark-streaming-kafka-0-10_2.13
Kinesis	spark-streaming-kinesis-asl_2.13 [Amazon Software License]

최신 목록은 Maven 저장소에서 지원되는 전체 소스와 아티팩트 목록을 확인하세요.

StreamingContext 초기화 (Initializing StreamingContext)

Spark Streaming 프로그램을 초기화하려면, 모든 Spark Streaming 기능의 메인 진입점인 StreamingContext 객체를 만들어야 해요.

StreamingContext 객체는 SparkContext 객체로부터 만들 수 있습니다.

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

sc = SparkContext(master, appName)
ssc = StreamingContext(sc, 1)

appName 파라미터는 클러스터 UI에 표시할 애플리케이션 이름이에요. master는 Spark 또는 YARN 클러스터 URL이거나, 로컬 모드로 실행하기 위한 특별한 "local[*]" 문자열입니다. 실제로 클러스터에서 실행할 때는 프로그램에 master를 하드코딩하기보다 애플리케이션을 spark-submit으로 실행해서 그쪽에서 받는 편이 좋아요. 다만 로컬 테스트나 유닛 테스트에서는 "local[*]"을 넘겨서(로컬 시스템의 코어 수를 감지해서) Spark Streaming을 프로세스 안에서 실행할 수 있습니다.

배치 간격은 애플리케이션의 지연 요구사항과 사용 가능한 클러스터 리소스를 기준으로 정해야 해요. 자세한 내용은 Performance Tuning 섹션을 참고하세요.

StreamingContext 객체는 SparkConf 객체로부터 만들 수 있습니다.

import org.apache.spark._
import org.apache.spark.streaming._

val conf = new SparkConf().setAppName(appName).setMaster(master)
val ssc = new StreamingContext(conf, Seconds(1))

appName 파라미터는 클러스터 UI에 표시할 애플리케이션 이름이에요. master는 Spark, Kubernetes 또는 YARN 클러스터 URL이거나, 로컬 모드로 실행하기 위한 특별한 "local[*]" 문자열입니다. 실제로 클러스터에서 실행할 때는 프로그램에 master를 하드코딩하기보다 애플리케이션을 spark-submit으로 실행해서 그쪽에서 받는 편이 좋아요. 다만 로컬 테스트나 유닛 테스트에서는 "local[*]"을 넘겨서 Spark Streaming을 프로세스 안에서 실행할 수 있습니다. 참고로 이렇게 하면 내부적으로 모든 Spark 기능의 시작점인 SparkContext가 생성되고, ssc.sparkContext로 접근할 수 있어요.

배치 간격은 애플리케이션의 지연 요구사항과 사용 가능한 클러스터 리소스를 기준으로 정해야 해요. 자세한 내용은 Performance Tuning 섹션을 참고하세요.

StreamingContext 객체는 기존 SparkContext 객체로부터 만들 수도 있어요.

import org.apache.spark.streaming._

val sc = ...                // existing SparkContext
val ssc = new StreamingContext(sc, Seconds(1))

JavaStreamingContext 객체는 SparkConf 객체로부터 만들 수 있습니다.

import org.apache.spark.*;
import org.apache.spark.streaming.api.java.*;

SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
JavaStreamingContext ssc = new JavaStreamingContext(conf, new Duration(1000));

appName 파라미터는 클러스터 UI에 표시할 애플리케이션 이름이에요. master는 Spark 또는 YARN 클러스터 URL이거나, 로컬 모드로 실행하기 위한 특별한 "local[*]" 문자열입니다. 실제로 클러스터에서 실행할 때는 프로그램에 master를 하드코딩하기보다 애플리케이션을 spark-submit으로 실행해서 그쪽에서 받는 편이 좋아요. 다만 로컬 테스트나 유닛 테스트에서는 "local[*]"을 넘겨서 Spark Streaming을 프로세스 안에서 실행할 수 있습니다. 참고로 이렇게 하면 내부적으로 모든 Spark 기능의 시작점인 JavaSparkContext가 생성되고, ssc.sparkContext로 접근할 수 있어요.

배치 간격은 애플리케이션의 지연 요구사항과 사용 가능한 클러스터 리소스를 기준으로 정해야 해요. 자세한 내용은 Performance Tuning 섹션을 참고하세요.

JavaStreamingContext 객체는 기존 JavaSparkContext로부터 만들 수도 있어요.

import org.apache.spark.streaming.api.java.*;

JavaSparkContext sc = ...   //existing JavaSparkContext
JavaStreamingContext ssc = new JavaStreamingContext(sc, Durations.seconds(1));

컨텍스트를 정의한 뒤에는 다음 작업을 해야 합니다.

1. 입력 DStream을 만들어 입력 소스를 정의한다.
2. DStream에 변환과 출력 연산을 적용해 스트리밍 계산을 정의한다.
3. streamingContext.start()로 데이터 수신과 처리를 시작한다.
4. streamingContext.awaitTermination()으로 처리가 중지될 때(수동으로든 오류로든)까지 기다린다.
5. streamingContext.stop()으로 처리를 수동으로 중지할 수 있다.

기억할 점:

• 컨텍스트가 시작된 뒤에는 새 스트리밍 계산을 설정하거나 추가할 수 없어요.
• 컨텍스트가 중지된 뒤에는 다시 시작할 수 없습니다.
• JVM 안에는 한 번에 하나의 StreamingContext만 활성 상태일 수 있어요.
• StreamingContext의 stop()은 SparkContext도 함께 중지시킵니다. StreamingContext만 중지하려면 stop()의 선택 파라미터인 stopSparkContext를 false로 설정하세요.
• 이전 StreamingContext가 (SparkContext는 중지하지 않고) 중지된 상태라면, 같은 SparkContext를 재사용해 여러 StreamingContext를 만들 수 있어요.

이산화 스트림 (Discretized Streams, DStreams)

Discretized Stream(이산화 스트림) 또는 DStream은 Spark Streaming이 제공하는 기본 추상화예요. 소스에서 받은 입력 데이터 스트림일 수도, 입력 스트림을 변환해 만든 처리된 데이터 스트림일 수도 있는 연속적인 데이터 스트림을 나타내죠. 내부적으로 DStream은 불변의 분산 데이터셋이라는 Spark의 추상화인 RDD의 연속적인 시리즈로 표현됩니다(Spark Programming Guide 참고). DStream 안의 각 RDD는 특정 간격의 데이터를 담고 있어요.

DStream에 적용하는 어떤 연산이든 내부적으로는 기반이 되는 RDD들에 대한 연산으로 변환돼요. 앞서 줄 스트림을 단어로 바꾼 예시를 보면, lines DStream의 각 RDD에 flatMap 연산이 적용되어 words DStream의 RDD들이 생성돼요.

이런 내부 RDD 변환은 Spark 엔진이 계산합니다. DStream 연산은 이런 세부 내용 대부분을 숨겨주고, 개발자에게 더 편리한 고수준 API를 제공해요. 이 연산들에 대해서는 뒤에서 자세히 다룹니다.

입력 DStream과 리시버 (Input DStreams and Receivers)

입력 DStream은 스트리밍 소스에서 받은 입력 데이터의 스트림을 나타내는 DStream이에요. 앞선 퀵 예제에서 lines는 netcat 서버에서 받은 데이터 스트림을 나타냈으므로 입력 DStream이었죠. 모든 입력 DStream(이 섹션에서 나중에 다루는 파일 스트림 제외)은 Receiver(Scala 문서, Java 문서) 객체와 연결되는데, 이 객체가 소스에서 데이터를 받아 처리를 위해 Spark의 메모리에 저장해요.

Spark Streaming은 두 종류의 내장 스트리밍 소스를 제공합니다.

• 기본 소스 (Basic sources): StreamingContext API에서 바로 사용할 수 있는 소스. 예: 파일시스템, 소켓 연결.
• 고급 소스 (Advanced sources): Kafka, Kinesis 같은 소스는 추가 유틸리티 클래스를 통해 사용할 수 있어요. 이들은 앞서 링킹 섹션에서 다룬 것처럼 추가 의존성과 링크해야 합니다.

각 카테고리의 소스 몇 가지를 이 섹션에서 다룰게요.

스트리밍 애플리케이션에서 여러 데이터 스트림을 병렬로 받고 싶다면, 입력 DStream을 여러 개 만들면 됩니다(성능 튜닝 섹션에서 더 설명할게요). 이렇게 하면 여러 리시버가 동시에 여러 데이터 스트림을 수신해요. 다만 Spark worker/executor는 오래 실행되는 태스크라서, Spark Streaming 애플리케이션에 할당된 코어 중 하나를 차지한다는 점을 주의해야 해요. 그래서 Spark Streaming 애플리케이션에는 받은 데이터를 처리하기에도, 리시버를 실행하기에도 충분한 코어(로컬로 실행하면 스레드)가 할당되어야 해요.

기억할 점

• 로컬로 Spark Streaming 프로그램을 실행할 때는 master URL에 "local"이나 "local[1]"을 쓰지 마세요. 둘 다 로컬에서 태스크를 실행할 때 스레드 하나만 쓰겠다는 뜻이에요. 리시버 기반 입력 DStream(소켓, Kafka 등)을 쓴다면 그 스레드 하나가 리시버를 실행하는 데 다 쓰여서, 받은 데이터를 처리할 스레드가 남지 않아요. 그래서 로컬 실행 시에는 항상 "local[n]"을 master URL로 쓰세요. 여기서 n은 실행할 리시버 수보다 커야 해요(master 설정 방법은 Spark Properties 참고).
• 클러스터에서 실행하는 경우로 확장하면, Spark Streaming 애플리케이션에 할당된 코어 수는 리시버 수보다 많아야 해요. 그렇지 않으면 시스템이 데이터는 받지만 처리하지 못해요.

기본 소스 (Basic Sources)

퀵 예제에서 TCP 소켓 연결로 받은 텍스트 데이터에서 DStream을 만드는 ssc.socketTextStream(...)을 이미 살펴봤어요. 소켓 외에도 StreamingContext API는 파일을 입력 소스로 삼아 DStream을 만드는 메서드를 제공합니다.

파일 스트림 (File Streams)

HDFS API와 호환되는(즉 HDFS, S3, NFS 등) 어떤 파일시스템에서든 파일을 읽으려면 StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass]로 DStream을 만들 수 있어요.

파일 스트림은 리시버를 실행할 필요가 없어서, 파일 데이터를 받기 위해 코어를 할당할 필요도 없습니다.

간단한 텍스트 파일을 다룰 때 가장 쉬운 방법은 StreamingContext.textFileStream(dataDirectory)예요.

fileStream은 Python API에서 사용할 수 없고, textFileStream만 사용할 수 있습니다.

streamingContext.textFileStream(dataDirectory)

streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)

텍스트 파일의 경우:

streamingContext.textFileStream(dataDirectory)

streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);

텍스트 파일의 경우:

streamingContext.textFileStream(dataDirectory);

디렉터리가 모니터링되는 방식 (How Directories are Monitored)

Spark Streaming은 dataDirectory 디렉터리를 모니터링하고, 그 디렉터리에 생성된 모든 파일을 처리해요.

• 단순 디렉터리를 모니터링할 수 있어요. 예: "hdfs://namenode:8040/logs/". 이런 경로 바로 아래의 모든 파일은 발견되는 대로 처리됩니다.
• POSIX glob 패턴을 지정할 수 있어요. 예: "hdfs://namenode:8040/logs/2017/*". 이 경우 DStream은 패턴과 일치하는 디렉터리의 모든 파일로 구성돼요. 즉, 이는 디렉터리 안의 파일 패턴이 아니라 디렉터리의 패턴이에요.
• 모든 파일은 같은 데이터 포맷이어야 합니다.
• 어떤 파일이 특정 시간 구간에 속하는지는 생성 시간이 아니라 수정 시간을 기준으로 판단해요.
• 한 번 처리된 파일은 현재 윈도우 안에서 내용이 바뀌어도 다시 읽히지 않아요. 즉, 업데이트는 무시됩니다.
• 디렉터리 아래 파일이 많을수록 변경 사항을 스캔하는 시간도 길어져요. 수정된 파일이 없더라도 마찬가지입니다.
• 와일드카드로 디렉터리를 식별할 때(예: "hdfs://namenode:8040/logs/2016-*"), 디렉터리 전체를 경로에 맞게 이름을 바꾸면 그 디렉터리가 모니터링 목록에 추가돼요. 현재 윈도우 안에 수정 시간이 있는 파일만 스트림에 포함됩니다.
• FileSystem.setTimes()를 호출해서 타임스탬프를 고치면, 내용이 안 바뀌었더라도 파일이 이후 윈도우에서 집히도록 할 수 있어요.

객체 스토어를 데이터 소스로 사용하기 (Using Object Stores as a source of data)

HDFS 같은 "완전한" 파일시스템은 출력 스트림이 생성되는 즉시 파일의 수정 시간을 설정하는 경향이 있어요. 파일이 완전히 쓰이기 전에 열리기라도 하면 그 파일이 DStream에 포함될 수 있고, 그러면 같은 윈도우 안에서의 파일 업데이트는 무시돼요. 즉 변경 사항을 놓치고, 스트림에서 데이터가 빠질 수 있습니다.

변경 사항이 윈도우에 반드시 잡히게 하려면, 파일을 모니터링하지 않는 디렉터리에 먼저 쓴 다음, 출력 스트림을 닫는 즉시 대상 디렉터리로 이름을 바꿔 옮기세요. 이름을 바꾼 파일이 생성된 윈도우 동안 스캔되는 대상 디렉터리에 나타나면, 새 데이터가 잡힙니다.

반대로 Amazon S3나 Azure Storage 같은 객체 스토어는 데이터가 실제로 복사되기 때문에 이름 바꾸기(rename) 연산이 보통 느려요. 게다가 이름을 바꾼 객체의 수정 시간이 rename() 연산 시점으로 설정될 수 있어서, 원래 생성 시간이 의미하던 윈도우에 포함되지 않을 수도 있어요.

대상 객체 스토어의 타임스탬프 동작이 Spark Streaming이 기대하는 것과 일치하는지, 그 스토어를 대상으로 조심스러운 테스트가 필요해요. 선택한 객체 스토어를 통해 스트리밍 데이터를 보낼 때는 대상 디렉터리에 직접 쓰는 전략이 적절할 수도 있어요.

이 주제에 대한 자세한 내용은 Hadoop Filesystem Specification을 참고하세요.

커스텀 리시버 기반 스트림 (Streams based on Custom Receivers)

커스텀 리시버로 받은 데이터 스트림으로 DStream을 만들 수 있어요. 자세한 내용은 Custom Receiver Guide를 참고하세요.

RDD 큐를 스트림으로 (Queue of RDDs as a Stream)

테스트 데이터로 Spark Streaming 애플리케이션을 테스트할 때는 streamingContext.queueStream(queueOfRDDs)로 RDD 큐에 기반한 DStream을 만들 수도 있어요. 큐에 들어가는 각 RDD는 DStream의 데이터 배치 하나로 취급되어 스트림처럼 처리됩니다.

소켓과 파일에서 오는 스트림에 대한 자세한 내용은 Python의 StreamingContext, Scala의 StreamingContext, Java의 JavaStreamingContext의 관련 함수 API 문서를 참고하세요.

고급 소스 (Advanced Sources)

Python API — Spark 4.2.0 기준으로 이 소스들 중 Kafka와 Kinesis는 Python API에서 사용 가능합니다.

이 카테고리의 소스는 외부의 Spark가 아닌 라이브러리와 연동이 필요하고, 그중 일부는 복잡한 의존성(예: Kafka)을 가져요. 그래서 의존성 버전 충돌 문제를 최소화하기 위해, 이 소스들로 DStream을 만드는 기능은 별도 라이브러리로 옮겨서 필요할 때 명시적으로 링크할 수 있게 했어요.

이 고급 소스들은 Spark shell에서는 사용할 수 없어서, 이들을 기반으로 한 애플리케이션은 shell에서 테스트할 수 없다는 점을 참고하세요. 정말로 Spark shell에서 쓰고 싶다면 해당 Maven 아티팩트의 JAR을 의존성과 함께 내려받아 클래스패스에 추가해야 합니다.

이 고급 소스 중 일부는 다음과 같습니다.

• Kafka: Spark Streaming 4.2.0은 Kafka 브로커 버전 0.10 이상과 호환돼요. 자세한 내용은 Kafka Integration Guide를 참고하세요.
• Kinesis: Spark Streaming 4.2.0은 Kinesis Client Library 1.2.1과 호환됩니다. 자세한 내용은 Kinesis Integration Guide를 참고하세요.

커스텀 소스 (Custom Sources)

Python API — Python에서는 아직 지원되지 않습니다.

커스텀 데이터 소스로도 입력 DStream을 만들 수 있어요. 커스텀 소스에서 데이터를 받아 Spark로 밀어넣는 사용자 정의 receiver(무엇인지는 다음 섹션에서 설명할게요)를 구현하기만 하면 됩니다. 자세한 내용은 Custom Receiver Guide를 참고하세요.

리시버 신뢰성 (Receiver Reliability)

데이터 소스는 *신뢰성(reliability)*에 따라 두 종류로 나뉠 수 있어요. (Kafka 같은) 소스는 전송된 데이터를 확인(acknowledgment)받을 수 있습니다. 이런 신뢰할 수 있는 소스에서 데이터를 받는 시스템이 받은 데이터를 올바르게 확인하면, 어떤 종류의 실패에도 데이터가 손실되지 않도록 보장할 수 있어요. 이로 인해 두 종류의 리시버가 생깁니다.

1. Reliable Receiver(신뢰할 수 있는 리시버) – 데이터가 Spark에 복제본과 함께 수신·저장된 뒤에야 신뢰할 수 있는 소스에 확인을 올바르게 보내는 리시버.
2. Unreliable Receiver(신뢰할 수 없는 리시버) – 소스에 확인을 보내지 않는 리시버. 확인을 지원하지 않는 소스에 쓸 수도 있고, 확인의 복잡성을 감수하고 싶지 않을 때 신뢰할 수 있는 소스에도 쓸 수 있어요.

신뢰할 수 있는 리시버를 만드는 방법에 대한 자세한 내용은 Custom Receiver Guide에서 다룹니다.

DStream 변환 (Transformations on DStreams)

RDD의 변환과 마찬가지로, 변환을 통해 입력 DStream의 데이터를 수정할 수 있어요. DStream은 일반적인 Spark RDD에서 사용할 수 있는 많은 변환을 지원합니다. 흔히 쓰는 것 몇 가지는 다음과 같아요.

변환	의미
map(func)	소스 DStream의 각 요소를 함수 func에 통과시켜 새 DStream을 반환한다.
flatMap(func)	map과 비슷하지만, 각 입력 항목을 출력 항목 0개 이상으로 매핑할 수 있다.
filter(func)	func가 true를 반환하는 소스 DStream의 레코드만 선택해 새 DStream을 반환한다.
repartition(numPartitions)	파티션을 늘리거나 줄여서 이 DStream의 병렬 처리 수준을 바꾼다.
union(otherStream)	소스 DStream과 otherDStream의 요소 합집합을 담은 새 DStream을 반환한다.
count()	소스 DStream의 각 RDD에 있는 요소 수를 세어 단일 요소 RDD들의 새 DStream을 반환한다.
reduce(func)	함수 func(인자 두 개를 받아 하나를 반환)로 소스 DStream의 각 RDD 요소를 집계해 단일 요소 RDD들의 새 DStream을 반환한다. 이 함수는 병렬로 계산될 수 있도록 결합 법칙과 교환 법칙을 만족해야 한다.
countByValue()	K 타입 요소의 DStream에서 호출하면, 각 키의 값이 소스 DStream의 각 RDD에서의 빈도인 (K, Long) 쌍의 새 DStream을 반환한다.
reduceByKey(func, [numTasks])	(K, V) 쌍의 DStream에서 호출하면, 각 키에 대해 주어진 reduce 함수로 값을 집계한 (K, V) 쌍의 새 DStream을 반환한다. 참고: 기본적으로 그룹화에는 Spark의 기본 병렬 태스크 수(로컬 모드에서는 2, 클러스터 모드에서는 spark.default.parallelism 설정 속성으로 정해짐)를 사용한다. 선택 인자 numTasks를 넘겨 태스크 수를 다르게 설정할 수 있다.
join(otherStream, [numTasks])	(K, V)와 (K, W) 쌍의 두 DStream에서 호출하면, 각 키에 대한 모든 요소 쌍을 가진 (K, (V, W)) 쌍의 새 DStream을 반환한다.
cogroup(otherStream, [numTasks])	(K, V)와 (K, W) 쌍의 DStream에서 호출하면, (K, Seq[V], Seq[W]) 튜플의 새 DStream을 반환한다.
transform(func)	소스 DStream의 모든 RDD에 RDD-to-RDD 함수를 적용해 새 DStream을 반환한다. DStream에서 임의의 RDD 연산을 수행하는 데 쓸 수 있다.
updateStateByKey(func)	각 키의 상태를, 그 키의 이전 상태와 키의 새 값들에 주어진 함수를 적용해 갱신하는 새 "state" DStream을 반환한다. 각 키에 대해 임의의 상태 데이터를 유지하는 데 쓸 수 있다.

이 변환 중 몇 가지는 더 자세히 살펴볼 가치가 있어요.

UpdateStateByKey 연산 (UpdateStateByKey Operation)

updateStateByKey 연산을 사용하면, 새 정보로 계속 갱신하면서 임의의 상태를 유지할 수 있어요. 이걸 쓰려면 두 단계가 필요합니다.

1. 상태를 정의한다 – 상태는 임의의 데이터 타입일 수 있어요.
2. 상태 갱신 함수를 정의한다 – 이전 상태와 입력 스트림에서 온 새 값들을 사용해 상태를 어떻게 갱신할지 함수로 지정한다.

매 배치마다 Spark는 (그 배치에 새 데이터가 있든 없든) 모든 기존 키에 대해 상태 갱신 함수를 적용해요. 갱신 함수가 None을 반환하면 그 키-값 쌍은 제거됩니다.

예시로 설명해볼게요. 텍스트 데이터 스트림에서 본 각 단어의 누적 카운트를 유지한다고 해볼게요. 여기서 누적 카운트가 상태이고, 정수예요. 갱신 함수를 이렇게 정의합니다.

def updateFunction(newValues, runningCount):
    if runningCount is None:
        runningCount = 0
    return sum(newValues, runningCount)  # add the new values with the previous running count to get the new count

이 값을 (앞선 예제의 (word, 1) 쌍을 담은 pairs DStream 같은) 단어를 담은 DStream에 적용해요.

runningCounts = pairs.updateStateByKey(updateFunction)

갱신 함수는 각 단어마다 호출되는데, newValues에는 1들의 시퀀스((word, 1) 쌍에서), runningCount에는 이전 카운트가 들어와요. 전체 Python 코드는 예제 stateful_network_wordcount.py를 참고하세요.

def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val newCount = ...  // add the new values with the previous running count to get the new count
    Some(newCount)
}

이 값을 (앞선 예제의 (word, 1) 쌍을 담은 pairs DStream 같은) 단어를 담은 DStream에 적용해요.

val runningCounts = pairs.updateStateByKey[Int](updateFunction _)

갱신 함수는 각 단어마다 호출되는데, newValues에는 1들의 시퀀스((word, 1) 쌍에서), runningCount에는 이전 카운트가 들어와요.

Function2<List<Integer>, Optional<Integer>, Optional<Integer>> updateFunction =
  (values, state) -> {
    Integer newSum = ...  // add the new values with the previous running count to get the new count
    return Optional.of(newSum);
  };

이 값을 (퀵 예제의 (word, 1) 쌍을 담은 pairs DStream 같은) 단어를 담은 DStream에 적용해요.

JavaPairDStream<String, Integer> runningCounts = pairs.updateStateByKey(updateFunction);

갱신 함수는 각 단어마다 호출되는데, newValues에는 1들의 시퀀스((word, 1) 쌍에서), runningCount에는 이전 카운트가 들어와요. 전체 Java 코드는 예제 JavaStatefulNetworkWordCount.java를 참고하세요.

updateStateByKey를 쓸 때는 checkpoint 디렉터리가 설정되어 있어야 한다는 점을 참고하세요. 자세한 내용은 checkpointing 섹션에서 다룹니다.

Transform 연산 (Transform Operation)

transform 연산(그 변형인 transformWith 등과 함께)은 DStream에 임의의 RDD-to-RDD 함수를 적용할 수 있게 해줘요. DStream API에 노출되지 않은 어떤 RDD 연산이든 적용하는 데 쓸 수 있습니다. 예를 들어 데이터 스트림의 매 배치를 다른 데이터셋과 조인하는 기능은 DStream API에 직접 노출되어 있지 않아요. 하지만 transform을 쓰면 쉽게 할 수 있죠. 이건 아주 강력한 가능성을 열어줍니다. 예를 들어 미리 계산한 스팸 정보(Spark로 만든 것일 수도 있어요)와 입력 데이터 스트림을 조인한 다음 그 기준으로 필터링하면 실시간 데이터 정리를 할 수 있어요.

spamInfoRDD = sc.pickleFile(...)  # RDD containing spam information

# join data stream with spam information to do data cleaning
cleanedDStream = wordCounts.transform(lambda rdd: rdd.join(spamInfoRDD).filter(...))

val spamInfoRDD = ssc.sparkContext.newAPIHadoopRDD(...) // RDD containing spam information

val cleanedDStream = wordCounts.transform { rdd =>
  rdd.join(spamInfoRDD).filter(...) // join data stream with spam information to do data cleaning
  ...
}

import org.apache.spark.streaming.api.java.*;
// RDD containing spam information
JavaPairRDD<String, Double> spamInfoRDD = jssc.sparkContext().newAPIHadoopRDD(...);

JavaPairDStream<String, Integer> cleanedDStream = wordCounts.transform(rdd -> {
  rdd.join(spamInfoRDD).filter(...); // join data stream with spam information to do data cleaning
  ...
});

제공된 함수는 배치 간격마다 호출된다는 점을 참고하세요. 덕분에 시간에 따라 변하는 RDD 연산, 즉 RDD 연산·파티션 수·브로드캐스트 변수 등을 배치 사이에 바꿀 수 있어요.

윈도우 연산 (Window Operations)

Spark Streaming은 데이터의 슬라이딩 윈도우에 걸쳐 변환을 적용할 수 있게 해주는 *윈도우 계산(windowed computations)*도 제공해요.

그림에서 보듯, 윈도우가 소스 DStream 위를 슬라이드할 때마다 윈도우 안에 들어오는 소스 RDD들이 결합되고 연산이 적용되어 윈도우 DStream의 RDD들이 생성돼요. 이 특정 경우엔 마지막 3개 시간 단위의 데이터에 연산을 적용하고, 2개 시간 단위씩 슬라이드합니다. 이것은 어떤 윈도우 연산이든 두 파라미터를 지정해야 한다는 뜻이에요.

• 윈도우 길이(window length) – 윈도우의 지속 시간(그림에서 3).
• 슬라이딩 간격(sliding interval) – 윈도우 연산이 수행되는 간격(그림에서 2).

이 두 파라미터는 소스 DStream의 배치 간격(그림에서 1)의 배수여야 합니다.

윈도우 연산을 예시로 설명해볼게요. 앞선 예제를 확장해서, 10초마다 최근 30초의 데이터로 단어 수를 생성한다고 해볼게요. 이렇게 하려면 지난 30초의 (word, 1) 쌍 pairs DStream에 reduceByKey 연산을 적용해야 해요. 이건 reduceByKeyAndWindow 연산으로 합니다.

# Reduce last 30 seconds of data, every 10 seconds
windowedWordCounts = pairs.reduceByKeyAndWindow(lambda x, y: x + y, lambda x, y: x - y, 30, 10)

// Reduce last 30 seconds of data, every 10 seconds
val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(30), Seconds(10))

// Reduce last 30 seconds of data, every 10 seconds
JavaPairDStream<String, Integer> windowedWordCounts = pairs.reduceByKeyAndWindow((i1, i2) -> i1 + i2, Durations.seconds(30), Durations.seconds(10));

흔히 쓰는 윈도우 연산 중 일부는 다음과 같아요. 이 연산들은 모두 앞서 말한 두 파라미터, windowLength와 slideInterval을 받습니다.

변환	의미
window(windowLength, slideInterval)	소스 DStream의 윈도우 배치를 기반으로 계산된 새 DStream을 반환한다.
countByWindow(windowLength, slideInterval)	스트림에서 요소의 슬라이딩 윈도우 카운트를 반환한다.
reduceByWindow(func, windowLength, slideInterval)	func로 슬라이딩 간격 동안 스트림의 요소를 집계해 만든 새 단일 요소 스트림을 반환한다. 이 함수는 병렬로 올바르게 계산될 수 있도록 결합 법칙과 교환 법칙을 만족해야 한다.
reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])	(K, V) 쌍의 DStream에서 호출하면, 슬라이딩 윈도우 안의 배치들에 주어진 reduce 함수 func로 각 키의 값을 집계한 (K, V) 쌍의 새 DStream을 반환한다. 참고: 기본적으로 그룹화에는 Spark의 기본 병렬 태스크 수(로컬 모드에서는 2, 클러스터 모드에서는 spark.default.parallelism 설정 속성으로 정해짐)를 사용한다. 선택 인자 numTasks를 넘겨 태스크 수를 다르게 설정할 수 있다.
reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])	위 reduceByKeyAndWindow()의 더 효율적인 버전으로, 각 윈도우의 reduce 값이 이전 윈도우의 reduce 값을 사용해 점진적으로 계산된다. 슬라이딩 윈도우에 새로 들어오는 데이터를 reduce하고, 윈도우를 떠나는 옛 데이터를 "역reduce(inverse reduce)"하는 방식으로 이룬다. 예를 들면 윈도우가 슬라이드할 때 키의 카운트를 "더하고""빼는" 방식이다. 다만 "역reduce 함수"(파라미터 invFunc로 받음)가 존재하는 "가역 reduce 함수(invertible reduce functions)"에만 적용할 수 있다. reduceByKeyAndWindow와 마찬가지로 reduce 태스크 수는 선택 인자로 설정 가능하다. 이 연산을 쓰려면 checkpointing이 활성화되어 있어야 한다는 점을 참고하자.
countByValueAndWindow(windowLength, slideInterval, [numTasks])	(K, V) 쌍의 DStream에서 호출하면, 각 키의 값이 슬라이딩 윈도우 안에서의 빈도인 (K, Long) 쌍의 새 DStream을 반환한다. reduceByKeyAndWindow와 마찬가지로 reduce 태스크 수는 선택 인자로 설정 가능하다.

조인 연산 (Join Operations)

마지막으로, Spark Streaming에서 다양한 종류의 조인을 얼마나 쉽게 수행할 수 있는지 짚고 넘어갈게요.

스트림-스트림 조인 (Stream-stream joins)

스트림은 다른 스트림과 아주 쉽게 조인할 수 있어요.

stream1 = ...
stream2 = ...
joinedStream = stream1.join(stream2)

val stream1: DStream[String, String] = ...
val stream2: DStream[String, String] = ...
val joinedStream = stream1.join(stream2)

JavaPairDStream<String, String> stream1 = ...
JavaPairDStream<String, String> stream2 = ...
JavaPairDStream<String, Tuple2<String, String>> joinedStream = stream1.join(stream2);

여기서 각 배치 간격마다 stream1이 생성한 RDD가 stream2가 생성한 RDD와 조인돼요. leftOuterJoin, rightOuterJoin, fullOuterJoin도 할 수 있죠. 게다가 스트림의 윈도우에 걸쳐 조인하는 것도 자주 유용한데, 그것도 꽤 쉽습니다.

windowedStream1 = stream1.window(20)
windowedStream2 = stream2.window(60)
joinedStream = windowedStream1.join(windowedStream2)

val windowedStream1 = stream1.window(Seconds(20))
val windowedStream2 = stream2.window(Minutes(1))
val joinedStream = windowedStream1.join(windowedStream2)

JavaPairDStream<String, String> windowedStream1 = stream1.window(Durations.seconds(20));
JavaPairDStream<String, String> windowedStream2 = stream2.window(Durations.minutes(1));
JavaPairDStream<String, Tuple2<String, String>> joinedStream = windowedStream1.join(windowedStream2);

스트림-데이터셋 조인 (Stream-dataset joins)

이건 앞서 DStream.transform 연산을 설명하면서 이미 봤어요. 윈도우 스트림을 데이터셋과 조인하는 또 다른 예시가 여기 있어요.

dataset = ... # some RDD
windowedStream = stream.window(20)
joinedStream = windowedStream.transform(lambda rdd: rdd.join(dataset))

val dataset: RDD[String, String] = ...
val windowedStream = stream.window(Seconds(20))...
val joinedStream = windowedStream.transform { rdd => rdd.join(dataset) }

JavaPairRDD<String, String> dataset = ...
JavaPairDStream<String, String> windowedStream = stream.window(Durations.seconds(20));
JavaPairDStream<String, String> joinedStream = windowedStream.transform(rdd -> rdd.join(dataset));

실은 조인할 데이터셋을 동적으로 바꿀 수도 있어요. transform에 제공된 함수는 배치 간격마다 평가되므로, 그 시점에 dataset 참조가 가리키는 현재 데이터셋을 사용해요.

DStream 변환의 전체 목록은 API 문서에서 확인할 수 있어요. Python API는 DStream, Scala API는 DStream과 PairDStreamFunctions, Java API는 JavaDStream과 JavaPairDStream을 참고하세요.

DStream 출력 연산 (Output Operations on DStreams)

출력 연산은 DStream의 데이터를 데이터베이스나 파일시스템 같은 외부 시스템으로 밀어내는 데 쓰여요. 출력 연산은 실제로 변환된 데이터를 외부 시스템이 소비하도록 하기 때문에, (RDD의 액션이 그런 것처럼) 모든 DStream 변환의 실제 실행을 촉발합니다. 현재 정의된 출력 연산은 다음과 같아요.

출력 연산	의미
print()	스트리밍 애플리케이션을 실행 중인 드라이버 노드에서 DStream의 각 배치 데이터의 처음 10개 요소를 출력한다. 개발과 디버깅에 유용하다. Python API — Python API에서는 **pprint()**라고 부른다.
saveAsTextFiles(prefix, [suffix])	이 DStream의 내용을 텍스트 파일로 저장한다. 각 배치 간격의 파일 이름은 prefix와 suffix를 기반으로 생성된다: "prefix-TIME_IN_MS[.suffix]".
saveAsObjectFiles(prefix, [suffix])	이 DStream의 내용을 직렬화된 Java 객체의 SequenceFiles로 저장한다. 각 배치 간격의 파일 이름은 prefix와 suffix를 기반으로 생성된다: "prefix-TIME_IN_MS[.suffix]". Python API — Python API에서는 사용할 수 없다.
saveAsHadoopFiles(prefix, [suffix])	이 DStream의 내용을 Hadoop 파일로 저장한다. 각 배치 간격의 파일 이름은 prefix와 suffix를 기반으로 생성된다: "prefix-TIME_IN_MS[.suffix]". Python API — Python API에서는 사용할 수 없다.
foreachRDD(func)	스트림에서 생성된 각 RDD에 함수 func를 적용하는 가장 범용적인 출력 연산자. 이 함수는 각 RDD의 데이터를 외부 시스템으로 밀어내야 하는데, 예를 들면 RDD를 파일로 저장하거나 네트워크를 통해 데이터베이스에 쓰는 식이다. 함수 func는 스트리밍 애플리케이션을 실행하는 드라이버 프로세스에서 실행되며, 보통 스트리밍 RDD의 계산을 강제하는 RDD 액션을 그 안에 담고 있다는 점을 참고하자.

foreachRDD 사용 패턴 (Design Patterns for using foreachRDD)

dstream.foreachRDD는 데이터를 외부 시스템으로 보낼 수 있게 해주는 강력한 기본 요소예요. 다만 이 기본 요소를 올바르고 효율적으로 사용하는 방법을 이해하는 게 중요합니다. 피해야 할 흔한 실수 몇 가지는 다음과 같아요.

외부 시스템에 데이터를 쓸 때는 종종 연결 객체(예: 원격 서버로의 TCP 연결)를 만들어야 해요. 이걸 매 레코드마다 만들면 과도한 오버헤드가 생길 수 있으니, 연결을 만들고 재사용하는 방법을 신중히 설계해야 해요. foreachRDD 안의 코드는 드라이버에서 실행되므로, 연결이나 자원은 RDD의 파티션 수준에서 관리하는 것이 효율적이에요.

체크포인팅 (Checkpointing)

• 메타데이터 체크포인팅 (Metadata checkpointing) – 스트리밍 계산을 정의하는 정보를 HDFS 같은 결함 허용 스토리지에 저장하는 것. 스트리밍 애플리케이션의 드라이버를 실행하는 노드가 실패했을 때 복구하는 데 사용된다(뒤에서 자세히 설명). 메타데이터에는 다음이 포함돼요:

  • 스트리밍 애플리케이션의 설정(구성)
  • DStream 연산의 정의 집합
  • 아직 완료되지 않은 배치들의 대기열

• 데이터 체크포인팅 (Data checkpointing) – 생성된 RDD를 신뢰할 수 있는 스토리지에 저장하는 것. (상태ful 변환, 즉 updateStateByKey나 슬라이딩 윈도우 같은) 상태 간격이 필요한 일부 상태ful 변환에서, 체크포인팅은 RDD들을 단계 간에 손실 없이 넘겨주고 결함 복구를 보장하는 데 필요해요. 데이터 체크포인팅은 저장된 데이터가 손실될 수 있어 이 걸리는 비용 때문에 상태ful 변환을 사용할 때만 설정하는 것이 좋아요.

애플리케이션 배포 (Deploying Applications)

요구사항 (Requirements)

• 클러스터 매니저가 있는 클러스터 (Cluster with a cluster manager) – 이것은 어떤 Spark 애플리케이션의 일반적인 요구사항이며, 배포 가이드에서 자세히 설명합니다.

• 애플리케이션 JAR 패키징 (Package the application JAR) – 스트리밍 애플리케이션을 JAR로 컴파일해야 해요. spark-submit로 애플리케이션을 시작한다면 JAR에 스트리밍 애플리케이션 코드뿐 아니라 내가 참조하는 모든 라이브러리도 포함되어야 합니다.

• 충분한 메모리 할당 (Configuring sufficient memory) – 수신된 데이터와 처리 중인 데이터를 저장하려면 클러스터가 스트리밍 애플리케이션에 충분한 메모리를 할당해야 해요.

• 체크포인팅 구성 (Configuring checkpointing) – 스트리밍 애플리케이션이 요구한다면, Hadoop API 호환의 결함 허용 스토리지(예: HDFS, S3 등)의 디렉터리를 체크포인트 디렉터리로 구성하고, 체크포인트 정보가 실패 복구에 사용될 수 있도록 스트리밍 애플리케이션을 작성해야 해요.

• YARN – YARN은 애플리케이션을 자동으로 다시 시작하는 비슷한 메커니즘을 지원해요. 자세한 내용은 YARN 문서를 참고하세요.

• 쓰기 전 로그 구성 (Configuring write-ahead logs) – Spark 1.2부터 강력한 결함 허용 보장을 위해 write-ahead logs를 도입했어요. S3(또는 플러시를 지원하지 않는 파일시스템)를 write-ahead logs에 사용한다면 spark.streaming.driver.writeAheadLog.closeFileAfterWrite와 spark.streaming.receiver.writeAheadLog.closeFileAfterWrite를 활성화하는 것을 잊지 마세요. 자세한 내용은 Spark Streaming Configuration을 참고하세요.

하지만 리시버 기반 입력의 경우 실패 시 데이터가 손실될 가능성은 여전히 존재해요. 그래서 리시버 기반 입력에는 좋은 결함 허용 의미를 제공하기 위해 복제와 쓰기 전 로그를 함께 사용하는 방법이 필요해요.

Spark 1.5에서 이러한 rate limit을 설정할 필요를 없애주는 *백프레셔(backpressure)*라는 기능을 도입했어요. Spark Streaming이 rate limit을 자동으로 알아내고, 처리 조건이 바뀌면 동적으로 조정하기 때문이죠. 이 백프레셔 기능은 spark.streaming.backpressure.enabled를 true로 설정하면 활성화할 수 있어요.

성능 튜닝 (Performance Tuning)

스트리밍 애플리케이션의 성능을 높이려면, 먼저 배치 처리 시간을 줄여서 각 배치에서 필요한 리소스를 최소화하거나, 배치 간격을 정확하게 맞춰 클러스터가 수신 속도를 따라잡도록 해야 해요. 어느 쪽이 주 병목인지 확인하려면 멀티 배치에 걸쳐 클러스터 모니터링 UI를 지켜보는 것이 좋아요.

배치 처리 시간 줄이기 (Reducing the Batch Processing Times)

데이터 처리의 병렬 수준 (Level of Parallelism in Data Processing)

데이터 수신과 처리를 병렬로 하려면, 그룹화 연산(즉 reduceByKey, reduceByKeyAndWindow 같은 연산)이 받은 데이터 파티션보다 더 많은 태스크로 병렬화되도록 해야 해요. reduceByKey 같은 연산은 기본적으로 spark.default.parallelism 설정 속성에 정의된 병렬 수준을 사용해요. 병렬 수준을 인자로 넘기거나(PairDStreamFunctions 문서 참고), spark.default.parallelism 설정 속성을 바꿔 기본값을 변경할 수 있습니다.

데이터 직렬화 (Data Serialization)

직렬화 오버헤드를 줄이면 데이터 전송을 크게 단축할 수 있어요. Spark Streaming의 경우 다음 옵션이 특히 중요합니다.

• Kryo 직렬화 사용 (Enable Kryo serialization) – Kryo 직렬화를 켜면 직렬화된 크기를 줄이고 네트워크 전송량을 줄여줘요. Spark 튜닝 가이드에서 자세히 설명하는 대로 클래스 등록과 함께 사용하는 것이 좋아요.
• 입력 데이터 압축 (Compress input data) – spark.rdd.compress 설정을 true로 설정하면, 배치 처리를 위해 저장되는 입력 데이터와 RDD 파티션을 압축할 수 있어요. 이 설정은 압축 해제 시 CPU 비용이 들지만, 1GB/s 이상의 네트워크 I/O를 제공할 수 있는 여유가 있는 시스템에서 메모리 사용량을 줄여줘요.

태스크 시작 오버헤드 (Task Launching Overheads)

마이크로초 수준의 작업 수천 개를 수신 데이터로 초당 처리한다면, 태스크 시작 오버헤드가 상당할 수 있어요. 이를 줄이려면:

• 태스크 직렬화 크기를 줄여라: 클러스터 모드에서는 압축된 직렬화 형식을 사용하거나, 태스크 실행기를 최적화하고 map 파티션을 줄여서 태스크당 더 많은 작업을 넣어야 해요.
• 작업당 처리되는 파티션 수를 조정해 실행기에 실행하는 태스크 수를 줄여라.

설정 클러스터의 처리량 (Setting the right batch interval)

배치 처리 시간이 두 배치 간격의 절반 이하가 되도록 배치 간격을 설정하는 것이 좋아요. 그러면 스트리밍 애플리케이션이 수신 속도를 따라잡을 수 있어요. 배치 간격을 얻기 어려워서 유지하기 어려운 경우엔, 백프레셔 기능을 활성화해서 수신 속도를 동적으로 제한할 수 있어요.

메모리 튜닝 (Memory Tuning)

Spark 애플리케이션의 메모리 사용량과 GC 동작을 튜닝하는 것은 Tuning Guide에 매우 자세히 설명되어 있어요. 그 문서를 읽어보길 강력히 권합니다. 이 섹션에서는 Spark Streaming 애플리케이션 맥락에서 특히 관련된 몇 가지 튜닝 파라미터를 다룰게요.

• DStream의 지속 수준 (Persistence Level of DStreams) – 앞서 Data Serialization 섹션에서 말했듯, 입력 데이터와 RDD는 기본적으로 직렬화된 바이트로 지속되어요. 이는 역직렬화된 지속보다 메모리 사용량과 GC 오버헤드를 모두 줄여줍니다. Kryo 직렬화를 활성화하면 직렬화된 크기와 메모리 사용량을 더 줄일 수 있어요. 메모리 사용량은 압축(Spark 설정 spark.rdd.compress)으로 더 줄일 수 있지만, CPU 시간이 들죠.

결함 허용 의미론 (Fault-tolerance Semantics)

기본 의미론 (Basic Semantics)

이 시점에서 스트리밍 애플리케이션의 정확성과 결함 허용에 대한 질문이 하나 남아 있어요. 이제 이걸 다룰게요.

1. 데이터 수신 (Receiving the data) – 각 입력 소스는 서로 다른 보장을 제공해요. 이건 다음 하위 섹션에서 자세히 설명합니다.
2. 데이터 변환 (Transforming the data) – RDD가 제공하는 보장 덕분에, 수신된 모든 데이터는 정확히 한 번(exactly once) 처리돼요. 단, 데이터 수신 중 실패 시 일부 입력 데이터가 여러 번 중복될 수 있다는 점에 유의하세요.
3. 데이터 출력 (Pushing out the data) – 출력 연산은 기본적으로 적어도 한 번(at-least once) 의미론을 제공해요. 네트워크에 쓰는 작업이 일부 성공 후 실패하면 그 데이터가 다시 쓰일 수 있으니까요.

하지만 사용자는 자신만의 트랜잭션 메커니즘을 구현해 정확히 한 번 의미론을 달성할 수 있어요. 이건 이 섹션 뒤에서 더 자세히 설명합니다.

수신 데이터의 의미론 (Semantics of Received Data)

서로 다른 입력 소스는 적어도 한 번부터 정확히 한 번까지 다양한 보장을 제공해요. 자세한 내용은 해당 소스 문서를 참고하세요.

• 소켓처럼 명시적인 확인(acknowledgment) 메커니즘이 없는 소스에서 데이터를 받는 경우: 이런 리시버는 어떤 오류 후에는 부분적으로 계산된 상태가 남을 수 있어요. 따라서 데이터가 소스에서 손실되면, 다시 전송되지 않아요. 그래서 소켓 스트림의 의미론은 적어도 한 번이 아닌 **최대 한 번(at-most once)**에 더 가깝습니다.

• 파일 스트림의 경우: 파일 스트림은 정확히 한 번 처리 보장을 제공해요.

• Kafka 리시버 기반 스트림의 경우: Kafka에서 리시버로 받는 데이터는 적어도 한 번 의미론을 제공해요.

Kafka Direct API 사용 시 (With Kafka Direct API)

Spark 1.3에서 새 Kafka Direct API를 도입했어요. 이 API는 모든 Kafka 데이터가 Spark Streaming에 정확히 한 번 수신되도록 보장할 수 있습니다. 여기에 정확히 한 번 출력 연산까지 구현하면, 엔드투엔드 정확히 한 번 보장을 달성할 수 있어요.

여기서 어디로 갈까 (Where to Go from Here)

• 추가 가이드
  • Kafka Integration Guide
  • Kinesis Integration Guide
  • Custom Receiver Guide
• 서드파티 DStream 데이터 소스는 Third Party Projects에서 찾을 수 있어요.
• API 문서
  • Python 문서: StreamingContext와 DStream
  • Scala 문서: StreamingContext, DStream
  • Java 문서: JavaStreamingContext, JavaDStream, JavaPairDStream, KafkaUtils, KinesisUtils
• 더 많은 예제: Scala와 Java 예제
• Spark Streaming을 설명하는 Paper와 video