RDD (RDD Programming Guide)

Spark 애플리케이션의 큰 그림은 생각보다 단순해요. 모든 Spark 애플리케이션은 사용자의 main 함수를 실행하는 드라이버 프로그램(driver program) 으로 시작해서, 클러스터 위에서 각종 **병렬 연산(parallel operations)**을 수행합니다. 그리고 그 병렬 연산의 중심에 서 있는 추상화가 바로 **복원 가능한 분산 데이터셋, RDD(Resilient Distributed Dataset)**예요.

RDD는 클러스터의 노드들에 걸쳐 **파티션(partition)**으로 나뉘어 저장된 요소들의 모음이라서, 요소들을 병렬로 처리할 수 있어요. RDD는 Hadoop 파일 시스템(또는 Hadoop이 지원하는 다른 파일 시스템)의 파일에서 만들거나, 드라이버 프로그램 안의 기존 Scala 컬렉션을 변환해서 만들 수 있어요. 그리고 사용자는 RDD를 메모리에 **영속화(persist)**할 수도 있는데, 그러면 여러 병렬 연산에 걸쳐 RDD를 효율적으로 재사용할 수 있죠. 마지막으로 RDD는 노드에 장애가 생겨도 자동으로 복구돼요.

Spark에는 또 하나의 추상화가 있는데, 바로 병렬 연산에 쓰이는 **공유 변수(shared variables)**예요. 기본적으로 Spark가 함수를 서로 다른 노드의 여러 태스크로 병렬 실행할 때, 함수에서 쓰는 각 변수의 복사본을 태스크마다 하나씩 보내줘요. 그런데 어떤 변수는 태스크끼리, 또는 태스크와 드라이버 프로그램 사이에서 공유돼야 할 때가 있어요. Spark는 이런 두 가지 유형의 공유 변수를 지원합니다. **브로드캐스트 변수(broadcast variables)**는 모든 노드의 메모리에 값을 캐시해 두는 데 쓰이고, **어큐뮬레이터(accumulators)**는 카운터나 합계처럼 "더하기만" 하는 변수예요.

이 가이드는 위 기능들을 Spark가 지원하는 각 언어에서 어떻게 쓰는지 보여줘요. 직접 따라 해 보려면 Spark의 대화형 셸을 띄우면 가장 좋아요. Scala 셸은 bin/spark-shell, Python 셸은 bin/pyspark를 실행하면 돼요.

Linking with Spark

Spark 4.2.0은 Python 3.10+와 함께 동작해요. 표준 CPython 인터프리터를 쓸 수 있어서 NumPy 같은 C 라이브러리도 그대로 쓸 수 있죠.

Python 애플리케이션은 런타임에 Spark를 포함하는 bin/spark-submit 스크립트로 실행하거나, setup.py에 다음과 같이 포함해 실행할 수 있어요.

    install_requires=[
        'pyspark==4.2.0'
    ]

pip으로 PySpark를 설치하지 않고 Python 애플리케이션을 실행하려면, Spark 디렉터리에 있는 bin/spark-submit 스크립트를 쓰면 돼요. 이 스크립트는 Spark의 Java/Scala 라이브러리를 로드해서 애플리케이션을 클러스터에 제출할 수 있게 해줘요. 대화형 Python 셸은 bin/pyspark로 실행할 수 있고요.

HDFS 데이터에 접근하려면, 사용 중인 HDFS 버전에 맞춰 빌드된 PySpark를 써야 해요. 흔한 HDFS 버전용 Prebuilt 패키지는 Spark 홈페이지에서 구할 수 있어요.

마지막으로 프로그램에 Spark 클래스들을 불러와야 해요. 다음 줄을 추가하면 됩니다.

from pyspark import SparkContext, SparkConf

PySpark는 드라이버와 워커 양쪽에서 같은 마이너 버전의 Python을 요구해요. 기본으로는 PATH에 있는 파이썬 버전을 쓰는데, PYSPARK_PYTHON으로 원하는 버전을 지정할 수 있어요. 예를 들면 이렇게요.

$ PYSPARK_PYTHON=python3.8 bin/pyspark

Spark 4.2.0은 기본적으로 Scala 2.13과 함께 빌드되어 배포돼요. (Spark는 다른 Scala 버전으로도 빌드할 수 있어요.) Scala로 애플리케이션을 쓰려면 호환되는 Scala 버전(예: 2.13.X)을 써야 합니다.

Scala 애플리케이션을 쓰려면 Maven 의존성에 Spark를 추가해야 해요. Spark는 Maven Central에서 구할 수 있어요.

groupId = org.apache.spark
artifactId = spark-core_2.13
version = 4.2.0

그리고 HDFS 클러스터에 접근해야 한다면, 사용 중인 HDFS 버전에 맞는 hadoop-client 의존성도 추가해야 해요.

groupId = org.apache.hadoop
artifactId = hadoop-client
version = <your-hdfs-version>

마지막으로 프로그램에 Spark 클래스들을 불러와야 해요. 다음 줄을 추가하면 됩니다.

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf

(Spark 1.3.0 이전에는 필수적인 암묵 변환(implicit conversions)을 활성화하려고 import org.apache.spark.SparkContext._ 를 명시적으로 써야 했어요.)

Spark 4.2.0은 함수를 간결하게 쓰는 람다 표현식(lambda expressions)을 지원해요. 아니면 org.apache.spark.api.java.function 패키지의 클래스들을 쓸 수도 있어요.

Java 7 지원은 Spark 2.2.0에서 제거됐다는 점을 참고하세요.

Java 애플리케이션을 쓰려면 Maven 의존성에 Spark를 추가해야 해요.

groupId = org.apache.spark
artifactId = spark-core_2.13
version = 4.2.0

HDFS 클러스터에 접근해야 한다면, 사용 중인 HDFS 버전의 hadoop-client 의존성도 추가해야 해요.

groupId = org.apache.hadoop
artifactId = hadoop-client
version = <your-hdfs-version>

마지막으로 프로그램에 Spark 클래스들을 불러와야 해요. 다음 줄을 추가하면 됩니다.

import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.SparkConf;

Initializing Spark

Spark 프로그램이 가장 먼저 해야 할 일은 SparkContext 객체를 만드는 거예요. 이 객체가 Spark에게 클러스터에 어떻게 접근할지 알려줘요. SparkContext를 만들려면 먼저 애플리케이션 정보가 담긴 SparkConf 객체를 만들어야 해요.

conf = SparkConf().setAppName(appName).setMaster(master)
sc = SparkContext(conf=conf)

Scala에서도 같아요. SparkContext 객체가 클러스터 접근 방식을 Spark에 알려주고, 만들기 전에 애플리케이션 정보가 담긴 SparkConf부터 만들어야 해요.

JVM 하나에는 SparkContext가 하나만 활성화돼야 해요. 새 SparkContext를 만들기 전에 기존 것을 반드시 stop() 해야 하죠.

val conf = new SparkConf().setAppName(appName).setMaster(master)
new SparkContext(conf)

Java는 JavaSparkContext를 써요. 역시 먼저 SparkConf를 만들고 나서 컨텍스트를 만듭니다.

SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
JavaSparkContext sc = new JavaSparkContext(conf);

appName 파라미터는 클러스터 UI에 표시할 애플리케이션 이름이에요. master는 Spark 또는 YARN 클러스터 URL이거나, 로컬 모드로 실행하기 위한 특수 문자열 "local"이에요. 실제로 클러스터에서 실행할 때는 프로그램에 master를 하드코딩하지 않고, spark-submit으로 애플리케이션을 실행하면서 받아오는 게 좋아요. 다만 로컬 테스트나 유닛 테스트에서는 "local"을 넘겨 프로세스 안에서 Spark를 실행할 수 있어요.

Using the Shell

PySpark 셸에는 인터프리터를 인식하는 SparkContext가 이미 만들어져 있어서, sc라는 변수로 쓸 수 있어요. 직접 SparkContext를 만들면 동작하지 않으니 주의하세요. 셸이 접속할 master는 --master 인자로 설정하고, Python .zip/.egg/.py 파일은 --py-files에 쉼표로 구분한 목록을 넘겨 런타임 경로에 추가할 수 있어요. 서드파티 Python 의존성은 Python Package Management를 참고하세요. Maven 좌표 목록을 --packages로 넘기면 셸 세션에 의존성(예: Spark Packages)을 추가할 수도 있어요. 의존성이 있을 수 있는 추가 저장소(예: Sonatype)는 --repositories로 넘겨요. 정확히 4개 코어로 bin/pyspark를 실행하려면 이렇게 하면 돼요.

$ ./bin/pyspark --master "local[4]"

나중에 import code할 수 있도록 code.py를 검색 경로에 추가하려면 이렇게 써요.

$ ./bin/pyspark --master "local[4]" --py-files code.py

옵션 전체 목록은 pyspark --help를 실행해 보세요. 사실 pyspark는 내부적으로 더 범용적인 spark-submit 스크립트를 호출해요.

PySpark 셸은 향상된 Python 인터프리터인 IPython에서도 띄울 수 있어요. PySpark는 IPython 1.0.0 이상에서 동작해요. IPython을 쓰려면 bin/pyspark 실행 시 PYSPARK_DRIVER_PYTHON 변수를 ipython으로 설정하면 됩니다.

$ PYSPARK_DRIVER_PYTHON=ipython ./bin/pyspark

Jupyter 노트북(이전의 IPython 노트북)을 쓰려면 이렇게 해요.

$ PYSPARK_DRIVER_PYTHON=jupyter PYSPARK_DRIVER_PYTHON_OPTS=notebook ./bin/pyspark

ipython 또는 jupyter 명령은 PYSPARK_DRIVER_PYTHON_OPTS로 커스터마이즈할 수 있어요.

Jupyter Notebook 서버가 시작된 뒤에는 "Files" 탭에서 새 노트북을 만들 수 있어요. 노트북 안에서 Spark를 시도하기 전에 %pylab inline 명령을 입력해 두면 됩니다.

Spark 셸에도 인터프리터를 인식하는 SparkContext가 sc라는 변수로 이미 만들어져 있어요. 직접 만들면 안 됩니다. master는 --master로, classpath에 추가할 JAR는 --jars에 쉼표로 구분한 목록으로 설정해요. Maven 좌표로 의존성(예: Spark Packages)을 추가하려면 --packages를, 추가 저장소(예: Sonatype)는 --repositories를 써요. 정확히 4개 코어로 bin/spark-shell을 실행하려면 이렇게 해요.

$ ./bin/spark-shell --master "local[4]"

classpath에 code.jar를 추가하려면 이렇게 해요.

$ ./bin/spark-shell --master "local[4]" --jars code.jar

Maven 좌표로 의존성을 포함하려면 이렇게 해요.

$ ./bin/spark-shell --master "local[4]" --packages "org.example:example:0.1"

옵션 전체 목록은 spark-shell --help를 실행해 보세요. spark-shell도 내부적으로 더 범용적인 spark-submit 스크립트를 호출해요.

Resilient Distributed Datasets (RDDs)

Spark는 **복원 가능한 분산 데이터셋(RDD)**이라는 개념을 중심으로 돌아가요. RDD는 병렬로 처리할 수 있는 장애 내성(fault-tolerant)을 가진 요소들의 모음이에요. RDD를 만드는 방법은 두 가지예요. 드라이버 프로그램에서 기존 컬렉션을 병렬화(parallelizing) 하거나, 공유 파일 시스템, HDFS, HBase, 또는 Hadoop InputFormat을 제공하는 어떤 데이터 소스 같은 외부 저장 시스템의 데이터셋을 참조하는 거죠.

Parallelized Collections

병렬화된 컬렉션은 드라이버 프로그램에서 기존 iterable(또는 컬렉션)에 SparkContext의 parallelize 메서드를 호출해서 만들어요. 컬렉션의 요소들은 복사되어 병렬로 처리할 수 있는 분산 데이터셋을 이룹니다. 예를 들어 1부터 5까지의 숫자를 담는 병렬화 컬렉션을 이렇게 만들어요.

data = [1, 2, 3, 4, 5]
distData = sc.parallelize(data)

만들어진 분산 데이터셋(distData)은 병렬로 처리할 수 있어요. 예를 들어 distData.reduce(lambda a, b: a + b)를 호출하면 리스트의 요소를 모두 더할 수 있죠. 분산 데이터셋의 연산은 뒤에서 더 설명할게요.

Scala에서는 드라이버 프로그램의 컬렉션(Scala Seq)에 parallelize를 호출해요.

val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)

만들어진 분산 데이터셋은 병렬로 처리할 수 있어요. distData.reduce((a, b) => a + b)로 배열의 요소를 더할 수 있죠.

Java에서는 JavaSparkContext의 parallelize 메서드를 기존 Collection에 호출해요.

List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
JavaRDD<Integer> distData = sc.parallelize(data);

병렬 컬렉션에서 중요한 파라미터 하나는 데이터를 몇 개로 쪼갤지 정하는 파티션(partitions) 수예요. Spark는 클러스터의 각 파티션마다 태스크 하나를 실행해요. 보통 클러스터의 CPU 하나당 2~4개의 파티션을 쓰는 게 좋아요. 기본적으로 Spark는 클러스터에 맞춰 파티션 수를 자동으로 정하려고 해요. 하지만 parallelize의 두 번째 파라미터로 직접 설정할 수도 있어요(예: sc.parallelize(data, 10)). 참고로 코드 일부에서는 하위 호환을 위해 파티션의 동의어인 slices라는 용어를 쓰기도 해요.

External Datasets

PySpark는 Hadoop이 지원하는 어떤 저장소로부터도 분산 데이터셋을 만들 수 있어요. 로컬 파일 시스템, HDFS, Cassandra, HBase, Amazon S3 등이 포함돼요. Spark는 텍스트 파일, SequenceFile, 그리고 그 밖의 모든 Hadoop InputFormat을 지원해요.

텍스트 파일 RDD는 SparkContext의 textFile 메서드로 만들어요. 이 메서드는 파일의 URI(로컬 경로 또는 hdfs://, s3a:// 등의 URI)를 받아서 그 파일을 **한 줄씩(line)**의 모음으로 읽어요. 호출 예시는 이렇습니다.

>>> distFile = sc.textFile("data.txt")

만들어진 distFile은 데이터셋 연산으로 다룰 수 있어요. 예를 들어 map과 reduce 연산으로 모든 줄의 크기를 더할 수 있죠: distFile.map(lambda s: len(s)).reduce(lambda a, b: a + b).

Spark로 파일을 읽을 때 몇 가지 알아둘 점이 있어요.

로컬 파일 시스템의 경로를 쓰는 경우, 그 파일은 워커 노드에서도 같은 경로로 접근 가능해야 해요. 파일을 모든 워커에 복사하거나, 네트워크로 마운트된 공유 파일 시스템을 써야 하죠.
textFile을 포함한 Spark의 모든 파일 기반 입력 메서드는 디렉터리, 압축 파일, 와일드카드에서도 동작해요. 예를 들어 textFile("/my/directory"), textFile("/my/directory/*.txt"), textFile("/my/directory/*.gz")처럼 쓸 수 있어요.
textFile 메서드는 파일의 파티션 수를 조절하는 선택적 두 번째 인자도 받아요. 기본적으로 Spark는 파일의 블록마다 파티션 하나를 만들어요(HDFS 기본값은 블록 128MB). 더 큰 값을 넘기면 더 많은 파티션을 요청할 수도 있지만, 블록 수보다 적은 파티션은 만들 수 없으니 참고하세요.

텍스트 파일 외에도 Spark의 Python API는 몇 가지 다른 데이터 포맷을 더 지원해요.

SparkContext.wholeTextFiles는 여러 작은 텍스트 파일이 든 디렉터리를 읽고, 각 파일을 (파일명, 내용) 쌍으로 반환해요. textFile이 파일마다 줄 단위 레코드를 하나씩 반환하는 것과 대비되죠.
RDD.saveAsPickleFile과 SparkContext.pickleFile은 pickle된 Python 객체로 이뤄진 단순한 포맷으로 RDD를 저장하고 로드해요. pickle 직렬화에는 batching이 쓰이며 기본 배치 크기는 10이에요.

SequenceFile 및 Hadoop Input/Output Format

이 기능은 현재 Experimental로 표시돼 있고 고급 사용자를 위한 것이에요. 나중에 Spark SQL 기반의 읽기/쓰기 지원으로 대체될 수 있는데, 그 경우 Spark SQL이 권장 방식이 돼요.

Writable 지원

PySpark의 SequenceFile 지원은 Java 내에서 키-값 쌍의 RDD를 로드하고, Writable을 기본 Java 타입으로 변환한 뒤 pickle로 결과 Java 객체를 pickle해요. 키-값 RDD를 SequenceFile로 저장할 때는 PySpark가 그 반대로 해요. Python 객체를 unpickle해서 Java 객체로 만들고, 다시 Writable로 변환하지요. 다음 Writable은 자동으로 변환돼요.

Writable 타입	Python 타입
Text	str
IntWritable	int
FloatWritable	float
DoubleWritable	float
BooleanWritable	bool
BytesWritable	bytearray
NullWritable	None
MapWritable	dict

배열(Array)은 기본 지원되지 않아요. 읽거나 쓸 때 사용자가 커스텀 ArrayWritable 하위 타입을 지정해야 해요. 쓸 때는 배열을 커스텀 ArrayWritable 하위 타입으로 변환하는 커스텀 변환기도 지정해야 하고요. 읽을 때 기본 변환기는 커스텀 ArrayWritable 하위 타입을 Java Object[]로 변환하며, 이건 pickle되어 Python 튜플이 돼요. 기본 타입의 배열로 Python array.array를 얻으려면 커스텀 변환기를 지정해야 합니다.

SequenceFile 저장 및 로드

텍스트 파일과 마찬가지로 경로를 지정하면 SequenceFile을 저장하고 로드할 수 있어요. 키와 값 클래스를 지정할 수 있지만, 표준 Writable이라면 굳이 필요 없어요.

>>> rdd = sc.parallelize(range(1, 4)).map(lambda x: (x, "a" * x))
>>> rdd.saveAsSequenceFile("path/to/file")
>>> sorted(sc.sequenceFile("path/to/file").collect())
[(1, u'a'), (2, u'aa'), (3, u'aaa')]

다른 Hadoop Input/Output Format 저장 및 로드

PySpark는 '신형'과 '구형' Hadoop MapReduce API 모두에서 어떤 Hadoop InputFormat이라도 읽고 어떤 Hadoop OutputFormat이라도 쓸 수 있어요. 필요하면 Hadoop 설정을 Python dict로 넘길 수 있어요. Elasticsearch ESInputFormat를 쓰는 예시는 이렇습니다.

$ ./bin/pyspark --jars /path/to/elasticsearch-hadoop.jar
>>> conf = {"es.resource" : "index/type"}  # assume Elasticsearch is running on localhost defaults
>>> rdd = sc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat",
                             "org.apache.hadoop.io.NullWritable",
                             "org.elasticsearch.hadoop.mr.LinkedMapWritable",
                             conf=conf)
>>> rdd.first()  # the result is a MapWritable that is converted to a Python dict
(u'Elasticsearch ID',
 {u'field1': True,
  u'field2': u'Some Text',
  u'field3': 12345})

InputFormat이 단순히 Hadoop 설정과 입력 경로에 의존하고, 키·값 클래스가 위 표대로 쉽게 변환된다면, 이런 방식이 잘 동작해요.

Cassandra/HBase 같은 곳에서 로드한 커스텀 직렬화 바이너리 데이터가 있다면, 먼저 Scala/Java 쪽에서 pickle의 pickler가 다룰 수 있는 형태로 데이터를 변환해야 해요. 이를 위한 Converter 트레이트가 제공돼요. 이 트레이트를 확장하고 convert 메서드에 변환 코드를 구현하면 됩니다. 이 클래스와, InputFormat 접근에 필요한 의존성들이 Spark job JAR에 패키징되어 PySpark classpath에 포함되도록 해야 해요.

Cassandra/HBase InputFormat·OutputFormat을 커스텀 변환기와 함께 쓰는 예시는 Python examples와 Converter examples에서 볼 수 있어요.

Scala에서도 Hadoop이 지원하는 모든 저장소에서 분산 데이터셋을 만들 수 있어요 — 로컬 파일 시스템, HDFS, Cassandra, HBase, Amazon S3 등. 텍스트 파일, SequenceFile, 그리고 그 밖의 Hadoop InputFormat을 지원해요.

텍스트 파일 RDD는 SparkContext의 textFile 메서드로 만들어요. 파일의 URI(로컬 경로 또는 hdfs://, s3a:// 등)를 받아 한 줄씩의 모음으로 읽죠. 호출 예시는 이렇습니다.

scala> val distFile = sc.textFile("data.txt")
distFile: org.apache.spark.rdd.RDD[String] = data.txt MapPartitionsRDD[10] at textFile at <console>:26

만들어진 distFile은 데이터셋 연산으로 다룰 수 있어요. map과 reduce로 모든 줄의 크기를 더할 수 있죠: distFile.map(s => s.length).reduce((a, b) => a + b).

Scala로 파일을 읽을 때도 앞서 본 주의점이 그대로 적용돼요.

로컬 파일 시스템 경로는 워커 노드에서도 같은 경로로 접근 가능해야 해요.
textFile을 포함한 모든 파일 기반 입력 메서드는 디렉터리, 압축 파일, 와일드카드에서 동작해요(예: textFile("/my/directory"), textFile("/my/directory/*.txt"), textFile("/my/directory/*.gz")). 여러 파일을 읽으면 파티션의 순서는 파일 시스템이 파일을 반환하는 순서에 따라 결정돼요. 경로의 사전식 순서를 따르거나 따르지 않을 수 있어요. 파티션 안에서는 요소가 기본 파일에서의 순서대로 정렬돼요.
textFile은 파티션 수를 조절하는 선택적 두 번째 인자를 받아요. 기본적으로 파일 블록마다 파티션 하나(HDFS 기본 블록 128MB)를 만들지만, 더 큰 값을 넘기면 더 많은 파티션을 요청할 수 있어요. 블록 수보다 적은 파티션은 만들 수 없어요.

텍스트 파일 외에도 Spark의 Scala API는 몇 가지 다른 데이터 포맷을 더 지원해요.

SparkContext.wholeTextFiles는 여러 작은 텍스트 파일이 든 디렉터리를 읽어 각 파일을 (파일명, 내용) 쌍으로 반환해요. textFile이 줄 단위 레코드를 반환하는 것과 대비되죠. 파티셔닝은 데이터 지역성(data locality)에 따라 결정되는데, 어떤 경우 파티션이 너무 적게 만들어질 수 있어요. 그럴 땐 wholeTextFiles가 최소 파티션 수를 조절하는 선택적 두 번째 인자를 제공해요.
SequenceFile은 SparkContext의 sequenceFile[K, V] 메서드를 써요. 여기서 K와 V는 파일의 키·값 타입이에요. 이들은 Hadoop의 Writable 인터페이스의 하위 클래스여야 해요(IntWritable, Text 등). 또한 Spark는 몇 가지 흔한 Writable에 대해 기본 타입을 지정할 수 있어요. 예를 들어 sequenceFile[Int, String]은 IntWritable과 Text를 자동으로 읽어요.
다른 Hadoop InputFormat은 SparkContext.hadoopRDD 메서드로 쓸 수 있어요. 임의의 JobConf, input format 클래스, 키 클래스, 값 클래스를 받아요. Hadoop job을 설정할 때와 똑같이 설정하면 돼요. "신형" MapReduce API(org.apache.hadoop.mapreduce) 기반 InputFormat에는 SparkContext.newAPIHadoopRDD를 쓸 수도 있어요.
RDD.saveAsObjectFile과 SparkContext.objectFile은 직렬화된 Java 객체로 이뤄진 단순한 포맷으로 RDD를 저장하고 로드해요. Avro처럼 특화된 포맷만큼 효율적이진 않지만, 어떤 RDD든 쉽게 저장할 수 있는 방법이에요.

Java도 Hadoop이 지원하는 모든 저장소에서 분산 데이터셋을 만들 수 있어요 — 로컬 파일 시스템, HDFS, Cassandra, HBase, Amazon S3 등. 텍스트 파일, SequenceFile, 그리고 그 밖의 Hadoop InputFormat을 지원해요.

텍스트 파일 RDD는 SparkContext의 textFile 메서드로 만들어요.

JavaRDD<String> distFile = sc.textFile("data.txt");

만들어진 distFile은 데이터셋 연산으로 다룰 수 있어요. map과 reduce로 모든 줄의 크기를 더할 수 있죠: distFile.map(s -> s.length()).reduce((a, b) -> a + b).

Java로 파일을 읽을 때 주의점은 같아요.

로컬 파일 시스템 경로는 워커 노드에서도 같은 경로로 접근 가능해야 해요.
모든 파일 기반 입력 메서드는 디렉터리, 압축 파일, 와일드카드에서 동작해요(예: textFile("/my/directory"), textFile("/my/directory/*.txt"), textFile("/my/directory/*.gz")).
textFile은 파티션 수를 조절하는 선택적 두 번째 인자를 받아요. 기본은 블록마다 파티션 하나(HDFS 기본 블록 128MB)지만, 더 큰 값을 넘기면 더 많은 파티션을 요청할 수 있어요. 블록 수보다 적은 파티션은 만들 수 없어요.

텍스트 파일 외에도 Spark의 Java API는 몇 가지 다른 데이터 포맷을 더 지원해요.

JavaSparkContext.wholeTextFiles는 여러 작은 텍스트 파일이 든 디렉터리를 읽어 (파일명, 내용) 쌍으로 반환해요. textFile이 줄 단위 레코드를 반환하는 것과 대비되죠.
SequenceFile은 SparkContext의 sequenceFile[K, V] 메서드를 써요. K와 V는 Hadoop의 Writable 인터페이스의 하위 클래스여야 해요(IntWritable, Text 등).
다른 Hadoop InputFormat은 JavaSparkContext.hadoopRDD 메서드로 쓸 수 있어요. 임의의 JobConf, input format 클래스, 키 클래스, 값 클래스를 받아요. Hadoop job 설정과 똑같이 설정하면 돼요. "신형" MapReduce API에는 JavaSparkContext.newAPIHadoopRDD를 쓸 수도 있어요.
JavaRDD.saveAsObjectFile과 JavaSparkContext.objectFile은 직렬화된 Java 객체로 이뤄진 단순한 포맷으로 RDD를 저장하고 로드해요.

RDD Operations

RDD는 두 종류의 연산을 지원해요. 트랜스포메이션(transformations) 은 기존 데이터셋에서 새 데이터셋을 만들고, 액션(actions) 은 데이터셋에서 계산을 실행한 뒤 값을 드라이버 프로그램에 반환해요. 예를 들어 map은 각 데이터셋 요소를 함수에 통과시켜 결과를 나타내는 새 RDD를 반환하는 트랜스포메이션이에요. 반대로 reduce는 어떤 함수로 RDD의 모든 요소를 집계해서 최종 결과를 드라이버 프로그램에 반환하는 액션이죠(분산 데이터셋을 반환하는 reduceByKey 같은 병렬 버전도 있긴 해요).

Spark의 모든 트랜스포메이션은 **lazy(지연)**해요. 즉 결과를 즉시 계산하지 않아요. 대신 어떤 기본 데이터셋(예: 파일)에 적용된 트랜스포메이션만 기억해 두죠. 실제 계산은 액션이 결과를 드라이버에 반환하려 할 때 비로소 일어나요. 이 설계 덕분에 Spark가 더 효율적으로 동작할 수 있어요. 예를 들어 map으로 만든 데이터셋이 reduce에서만 쓰일 거라면, 더 큰 map 결과 전체 대신 reduce의 결과만 드라이버로 보내면 되죠.

기본적으로 변환된 RDD는 액션을 실행할 때마다 다시 계산될 수 있어요. 하지만 persist(또는 cache) 메서드로 RDD를 메모리에 **영속화(persist)**할 수 있어요. 그러면 Spark가 요소를 클러스터에 보관해 두고, 다음에 조회할 때 훨씬 빠르게 접근해요. RDD를 디스크에 영속화하거나 여러 노드에 복제해 저장하는 것도 지원해요.

Basics

RDD의 기초를 보여주는 아주 간단한 프로그램을 볼게요.

lines = sc.textFile("data.txt")
lineLengths = lines.map(lambda s: len(s))
totalLength = lineLengths.reduce(lambda a, b: a + b)

첫 줄은 외부 파일에서 기본 RDD를 정의해요. 이 데이터셋은 메모리에 로드되지도 않고 당장 처리되지도 않아요. lines는 그저 파일을 가리키는 포인터일 뿐이죠. 둘째 줄은 map 트랜스포메이션의 결과로 lineLengths를 정의해요. 이것도 lazy 성질 때문에 즉시 계산되지 않아요. 마지막으로 reduce라는 액션을 실행해요. 이 시점에 Spark는 계산을 여러 머신에서 돌릴 작업들로 쪼개고, 각 머신이 자기 몫의 map과 로컬 reduce를 실행한 뒤 자기 답만 드라이버 프로그램에 반환해요.

나중에 lineLengths를 다시 쓰고 싶다면, reduce 앞에 이렇게 한 줄을 추가하면 돼요.

lineLengths.persist()

그러면 lineLengths가 처음 계산된 뒤 메모리에 저장돼요.

Scala로 보면 같은 흐름이에요.

val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)

lines는 파일 포인터일 뿐이고, lineLengths는 lazy하게 정의되며, reduce 액션을 만난 시점에 비로소 각 머신이 map과 로컬 reduce를 실행해 답만 돌려줘요. 나중에 다시 쓰려면 다음처럼 persist()를 reduce 앞에 넣으면 돼요.

lineLengths.persist()

Java도 똑같은 구조예요.

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(s -> s.length());
int totalLength = lineLengths.reduce((a, b) -> a + b);

Java에서는 persist 호출에 저장 레벨을 명시해야 해요.

lineLengths.persist(StorageLevel.MEMORY_ONLY());

Passing Functions to Spark

Spark의 API는 드라이버 프로그램에서 함수를 정의해 클러스터에서 실행하는 방식에 크게 의존해요. Python에서 함수를 넘기는 권장 방법은 세 가지예요.

람다 표현식(lambda expressions) — 표현식으로 쓸 수 있는 간단한 함수용. (람다는 여러 문장으로 된 함수나 값을 반환하지 않는 문장은 지원하지 않아요.)
더 긴 코드용으로는 Spark를 호출하는 함수 안의 로컬 def.
모듈의 최상위 함수.

lambda로는 감당하기 힘든 더 긴 함수를 넘겨야 한다면 이렇게 써요.

"""MyScript.py"""
if __name__ == "__main__":
    def myFunc(s):
        words = s.split(" ")
        return len(words)

    sc = SparkContext(...)
    sc.textFile("file.txt").map(myFunc)

클래스 인스턴스의 메서드(싱글턴 객체가 아니라) 참조를 넘기는 것도 가능하지만, 이 경우 메서드가 든 객체 자체를 함께 보내야 해요. 예를 들어 보면 이렇습니다.

class MyClass(object):
    def func(self, s):
        return s
    def doStuff(self, rdd):
        return rdd.map(self.func)

여기서 new MyClass를 만들고 doStuff를 호출하면, 안의 map은 그 MyClass 인스턴스의 func 메서드를 참조하므로 전체 객체가 클러스터로 보내져야 해요.

비슷하게, 바깥 객체의 필드에 접근하는 것도 전체 객체를 참조하게 돼요.

class MyClass(object):
    def __init__(self):
        self.field = "Hello"
    def doStuff(self, rdd):
        return rdd.map(lambda s: self.field + s)

이 문제를 피하는 가장 간단한 방법은 필드를 바깥에서 접근하지 말고 로컬 변수로 복사하는 거예요.

def doStuff(self, rdd):
    field = self.field
    return rdd.map(lambda s: field + s)

Spark의 API는 클러스터에서 실행할 함수를 드라이버에서 넘기는 데 크게 의존해요. Scala에서는 두 가지 방법을 권장해요.

짧은 코드에 쓸 수 있는 익명 함수 문법(anonymous function syntax).
전역 싱글턴 객체의 static 메서드. 예를 들어 object MyFunctions를 정의하고 MyFunctions.func1을 넘길 수 있어요.

object MyFunctions {
  def func1(s: String): String = { ... }
}

myRdd.map(MyFunctions.func1)

클래스 인스턴스의 메서드(싱글턴 객체가 아니라) 참조를 넘기는 것도 가능하지만, 메서드가 든 객체를 함께 보내야 해요.

class MyClass {
  def func1(s: String): String = { ... }
  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
}

여기서 MyClass 인스턴스를 만들고 doStuff를 호출하면, 안의 map은 그 MyClass 인스턴스의 func1 메서드를 참조하므로 전체 객체가 클러스터로 보내져요. rdd.map(x => this.func1(x))로 쓰는 것과 같죠.

바깥 객체의 필드에 접근하는 것도 전체 객체를 참조하게 돼요.

class MyClass {
  val field = "Hello"
  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
}

이건 rdd.map(x => this.field + x)로 쓰는 것과 같아서 this 전체를 참조해요. 피하는 가장 간단한 방법은 필드를 로컬 변수로 복사하는 거예요.

def doStuff(rdd: RDD[String]): RDD[String] = {
  val field_ = this.field
  rdd.map(x => field_ + x)
}

Spark의 API는 클러스터에서 실행할 함수를 드라이버에서 넘기는 데 의존해요. Java에서 함수는 org.apache.spark.api.java.function 패키지의 인터페이스를 구현하는 클래스로 표현돼요. 이런 함수를 만드는 방법은 두 가지예요.

함수 인터페이스를 자신의 클래스(익명 내부 클래스든 이름 있는 클래스든)로 구현하고 그 인스턴스를 Spark에 넘기기.
람다 표현식으로 구현을 간결하게 정의하기.

이 가이드의 대부분은 간결함을 위해 람다 문법을 쓰지만, 모든 API는 장황한(long-form) 방식으로도 똑같이 쓸 수 있어요. 위 코드를 다음과 같이 다르게 쓸 수도 있었죠.

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(new Function<String, Integer>() {
  public Integer call(String s) { return s.length(); }
});
int totalLength = lineLengths.reduce(new Function2<Integer, Integer, Integer>() {
  public Integer call(Integer a, Integer b) { return a + b; }
});

아니면 함수를 인라인으로 쓰는 게 지저분하다면 이름 있는 클래스로 분리해도 돼요.

class GetLength implements Function<String, Integer> {
  public Integer call(String s) { return s.length(); }
}
class Sum implements Function2<Integer, Integer, Integer> {
  public Integer call(Integer a, Integer b) { return a + b; }
}

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(new GetLength());
int totalLength = lineLengths.reduce(new Sum());

Java의 익명 내부 클래스는 final로 표시된 경우에만 둘러싼 스코프의 변수에 접근할 수 있어요. Spark는 다른 언어에서처럼 이런 변수들의 복사본을 각 워커 노드로 보내요.

Understanding closures

Spark에서 다소 어려운 부분 중 하나는, 코드가 클러스터에 걸쳐 실행될 때 변수와 메서드의 스코프와 생명 주기를 이해하는 거예요. 자기 스코프 바깥의 변수를 수정하는 RDD 연산은 혼란의 흔한 원인이 되죠. 아래 예시는 foreach()로 카운터를 증가시키는 코드를 살펴볼게요. 비슷한 문제는 다른 연산에서도 생길 수 있어요.

Example

아래의 단순한 RDD 요소 합계 코드를 봐요. 이 코드는 실행이 같은 JVM 안에서 일어나는지에 따라 다르게 동작할 수 있어요. 흔한 예는 Spark를 local 모드(--master = "local[n]")로 돌릴 때와 애플리케이션을 클러스터(spark-submit으로 YARN 등)에 배포할 때의 차이죠.

counter = 0
rdd = sc.parallelize(data)

# Wrong: Don't do this!!
def increment_counter(x):
    global counter
    counter += x
rdd.foreach(increment_counter)

print("Counter value: ", counter)

var counter = 0
var rdd = sc.parallelize(data)

// Wrong: Don't do this!!
rdd.foreach(x => counter += x)

println("Counter value: " + counter)

int counter = 0;
JavaRDD<Integer> rdd = sc.parallelize(data);

// Wrong: Don't do this!!
rdd.foreach(x -> counter += x);

println("Counter value: " + counter);

Local vs. cluster modes

위 코드의 동작은 정의되어 있지 않고, 의도대로 동작하지 않을 수 있어요. 작업을 실행하기 위해 Spark는 RDD 연산의 처리를 태스크들로 쪼개고, 각 태스크는 실행기(executor)가 실행해요. 실행 전에 Spark는 태스크의 **클로저(closure)**를 계산해요. 클로저란 실행기가 RDD(여기선 foreach())에 대한 계산을 수행하기 위해 보여야 하는 변수와 메서드의 집합이에요. 이 클로저는 직렬화되어 각 실행기로 보내진답니다.

각 실행기로 보내진 클로저 안의 변수는 복사본이에요. 그래서 foreach 함수 안에서 counter를 참조할 때, 그것은 더 이상 드라이버 노드의 counter가 아니에요. 드라이버 노드 메모리에는 여전히 counter가 있지만, 이제 실행기들이 그걸 볼 수 없죠! 실행기들은 직렬화된 클로저의 복사본만 봐요. 따라서 counter의 최종 값은 여전히 0이 돼요. 모든 연산이 직렬화된 클로저 안의 값을 참조했으니까요.

local 모드에서는 어떤 상황에서 foreach 함수가 드라이버와 같은 JVM 안에서 실제로 실행되어 원래 counter를 참조하고 실제로 갱신할 수도 있어요.

이런 시나리오에서 잘 정의된 동작을 보장하려면 Accumulator를 써야 해요. Spark의 Accumulator는 클러스터의 워커 노드에 걸쳐 실행이 쪼개질 때 변수를 안전하게 갱신하는 메커니즘을 제공하기 위해 특별히 쓰여요. Accumulator에 대한 자세한 내용은 이 가이드의 Accumulators 섹션에서 다룬답니다.

일반적으로 클로저(루프나 로컬로 정의된 메서드 같은 구조)로 어떤 전역 상태를 변경해서는 안 돼요. Spark는 클로저 바깥에서 참조된 객체의 변경 동작을 정의하거나 보장하지 않아요. 그런 코드는 local 모드에서 동작할 수도 있는데, 그건 그냥 우연일 뿐이라 분산 모드에서는 기대대로 동작하지 않을 거예요. 전역 집계가 필요하다면 Accumulator를 쓰세요.

Printing elements of an RDD

또 하나 흔한 패턴은 rdd.foreach(println) 또는 rdd.map(println)으로 RDD의 요소를 출력하려는 거예요. 단일 머신에서는 기대한 출력이 나와서 RDD의 모든 요소가 인쇄돼요. 하지만 cluster 모드에서는 실행기가 호출하는 stdout 출력이 실행기 자신의 stdout으로 가고, 드라이버의 stdout으론 가지 않아서 드라이버에서 볼 수 없어요! 모든 요소를 드라이버에 출력하려면 collect() 메서드로 RDD를 드라이버 노드로 가져온 뒤 rdd.collect().foreach(println)처럼 하면 돼요. 다만 collect()는 RDD 전체를 단일 머신으로 가져오므로 드라이버 메모리가 부족해질 수도 있어요. 요소 몇 개만 출력하면 된다면 take()를 쓰는 게 더 안전해요: rdd.take(100).foreach(println).

Working with Key-Value Pairs

대부분의 Spark 연산은 어떤 타입의 객체든 담은 RDD에서 동작하지만, 몇 가지 특별한 연산은 키-값 쌍 RDD에서만 쓸 수 있어요. 가장 흔한 건 키로 요소를 그룹화하거나 집계하는 분산 "셔플(shuffle)" 연산이에요.

Python에서는 이런 연산이 (1, 2) 같은 내장 Python 튜플을 담은 RDD에서 동작해요. 그런 튜플을 만들고 원하는 연산을 호출하기만 하면 됩니다.

예를 들어 다음 코드는 키-값 쌍에 reduceByKey 연산을 써서 파일에 각 줄이 몇 번 등장하는지 세요.

lines = sc.textFile("data.txt")
pairs = lines.map(lambda s: (s, 1))
counts = pairs.reduceByKey(lambda a, b: a + b)

예를 들어 counts.sortByKey()로 쌍을 알파벳순으로 정렬하고, 마지막엔 counts.collect()로 객체 리스트 형태로 드라이버 프로그램에 가져올 수 있어요.

Scala에서 이런 연산은 Tuple2 객체(그냥 (a, b)라고 쓰는 언어의 내장 튜플)를 담은 RDD에서 자동으로 쓸 수 있어요. 키-값 쌍 연산은 PairRDDFunctions 클래스에 있는데, 이 클래스가 튜플 RDD를 자동으로 감싸요.

val lines = sc.textFile("data.txt")
val pairs = lines.map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)

counts.sortByKey()로 쌍을 알파벳순 정렬하고, counts.collect()로 드라이버에 객체 배열로 가져올 수 있어요.

주의: 키-값 연산에서 커스텀 객체를 키로 쓸 때는, 커스텀 equals() 메서드에 짝이 맞는 hashCode() 메서드가 반드시 함께 있어야 해요. 자세한 내용은 Object.hashCode() 문서에 나온 계약을 참고하세요.

Java에서 키-값 쌍은 Scala 표준 라이브러리의 scala.Tuple2 클래스로 표현돼요. new Tuple2(a, b)로 튜플을 만들고, 나중에 tuple._1()과 tuple._2()로 필드에 접근할 수 있어요.

키-값 쌍 RDD는 JavaPairRDD 클래스로 표현돼요. mapToPair, flatMapToPair 같은 map 연산의 특별 버전으로 JavaRDD에서 JavaPairRDD를 만들 수 있어요. JavaPairRDD는 표준 RDD 함수와 특별한 키-값 함수를 모두 갖고 있어요.

JavaRDD<String> lines = sc.textFile("data.txt");
JavaPairRDD<String, Integer> pairs = lines.mapToPair(s -> new Tuple2(s, 1));
JavaPairRDD<String, Integer> counts = pairs.reduceByKey((a, b) -> a + b);

counts.sortByKey()로 알파벳순 정렬, counts.collect()로 드라이버에 배열로 가져올 수 있어요.

주의: 커스텀 객체를 키로 쓸 때는 equals()와 짝이 맞는 hashCode()가 함께 있어야 해요. 자세한 내용은 Object.hashCode() 문서를 참고하세요.

Transformations

다음 표는 Spark가 지원하는 흔한 트랜스포메이션 중 일부를 보여줘요. 자세한 내용은 RDD API 문서(Python, Scala, Java, R)와 Pair RDD 함수 문서(Scala, Java)를 참고하세요.

Transformation	의미
map(func)	소스의 각 요소를 함수 func에 통과시켜 만든 새 분산 데이터셋을 반환한다.
filter(func)	func가 true를 반환하는 소스 요소만 골라 만든 새 데이터셋을 반환한다.
flatMap(func)	map과 비슷하지만, 각 입력 항목이 0개 이상의 출력 항목으로 매핑될 수 있다(func는 단일 항목 대신 Seq를 반환해야 한다).
mapPartitions(func)	map과 비슷하지만 RDD의 각 파티션(블록)마다 개별적으로 실행된다. 타입 T의 RDD에서 실행하면 func는 `Iterator<T> => Iterator<U>` 타입이어야 한다.
mapPartitionsWithIndex(func)	mapPartitions와 비슷하지만 func에 파티션의 인덱스를 나타내는 정수 값을 추가로 제공한다. 타입 T의 RDD에서 실행하면 func는 `(Int, Iterator<T>) => Iterator<U>` 타입이어야 한다.
sample(withReplacement, fraction, seed)	데이터의 fraction 비율을, 주어진 난수 생성기 시드로 복원 추출 또는 비복원 추출로 샘플링한다.
union(otherDataset)	소스 데이터셋과 인자의 요소들의 합집합(union)을 담은 새 데이터셋을 반환한다.
intersection(otherDataset)	소스 데이터셋과 인자의 요소들의 교집합을 담은 새 RDD를 반환한다.
distinct([numPartitions])	소스 데이터셋의 고유 요소를 담은 새 데이터셋을 반환한다.
groupByKey([numPartitions])	(K, V) 쌍 데이터셋에서 호출하면 (K, Iterable<V>) 쌍 데이터셋을 반환한다. 주의: 각 키에 대해 집계(합이나 평균 등)를 하려고 그룹화한다면 `reduceByKey` 또는 `aggregateByKey`를 쓰는 게 성능이 훨씬 좋다. 주의: 기본적으로 출력의 병렬성 수준은 부모 RDD의 파티션 수에 달려 있다. 선택적 `numPartitions` 인자로 다른 태스크 수를 설정할 수 있다.
reduceByKey(func, [numPartitions])	(K, V) 쌍 데이터셋에서 호출하면, 주어진 reduce 함수 func(타입 `(V,V) => V`여야 함)로 각 키의 값을 집계한 (K, V) 쌍 데이터셋을 반환한다. `groupByKey`와 마찬가지로 reduce 태스크 수는 선택적 두 번째 인자로 설정할 수 있다.
aggregateByKey(zeroValue)(seqOp, combOp, [numPartitions])	(K, V) 쌍 데이터셋에서 호출하면, 주어진 결합 함수와 중립적 "zero" 값으로 각 키의 값을 집계한 (K, U) 쌍 데이터셋을 반환한다. 입력 값 타입과 다른 집계 값 타입을 허용하면서 불필요한 할당을 피한다. `groupByKey`와 마찬가지로 reduce 태스크 수는 선택적 두 번째 인자로 설정할 수 있다.
sortByKey([ascending], [numPartitions])	K가 Ordered를 구현하는 (K, V) 쌍 데이터셋에서 호출하면, 불리언 `ascending` 인자에 지정된 대로 키를 오름차순 또는 내림차순으로 정렬한 (K, V) 쌍 데이터셋을 반환한다.
join(otherDataset, [numPartitions])	(K, V)와 (K, W) 타입 데이터셋에서 호출하면, 각 키의 모든 요소 쌍을 담은 (K, (V, W)) 쌍 데이터셋을 반환한다. 외부 조인은 `leftOuterJoin`, `rightOuterJoin`, `fullOuterJoin`으로 지원된다.
cogroup(otherDataset, [numPartitions])	(K, V)와 (K, W) 타입 데이터셋에서 호출하면 (K, (Iterable<V>, Iterable<W>)) 튜플 데이터셋을 반환한다. 이 연산은 groupWith라고도 불린다.
cartesian(otherDataset)	타입 T와 U의 데이터셋에서 호출하면 (T, U) 쌍(모든 요소 쌍) 데이터셋을 반환한다.
pipe(command, [envVars])	RDD의 각 파티션을 셸 명령(예: Perl 또는 bash 스크립트)에 파이프한다. RDD 요소는 프로세스의 stdin으로 쓰이고, 그 stdout으로 출력된 줄들은 문자열 RDD로 반환된다.
coalesce(numPartitions)	RDD의 파티션 수를 numPartitions로 줄인다. 큰 데이터셋을 필터링한 뒤 더 효율적으로 연산을 실행할 때 유용하다.
repartition(numPartitions)	RDD의 데이터를 무작위로 다시 섞어 파티션을 늘리거나 줄이고 균형을 맞춘다. 항상 전체 데이터를 네트워크로 셔플한다.
repartitionAndSortWithinPartitions(partitioner)	주어진 partitioner에 따라 RDD를 재파티셔닝하고, 각 결과 파티션 안에서 레코드를 키로 정렬한다. repartition 후 각 파티션을 정렬하는 것보다 효율적인데, 셔플 메커니즘 안으로 정렬을 밀어 넣을 수 있기 때문이다.

Actions

다음 표는 Spark가 지원하는 흔한 액션 중 일부를 보여줘요. 자세한 내용은 RDD API 문서와 Pair RDD 함수 문서(Scala, Java)를 참고하세요.

Action	의미
reduce(func)	함수 func(인자 둘을 받아 하나를 반환)로 데이터셋의 요소를 집계한다. 함수는 병렬로 올바르게 계산될 수 있도록 교환·결합 법칙을 만족해야 한다.
collect()	데이터셋의 모든 요소를 드라이버 프로그램의 배열로 반환한다. 보통 filter 또는 충분히 작은 부분집합을 반환하는 연산 다음에 유용하다.
count()	데이터셋의 요소 개수를 반환한다.
first()	데이터셋의 첫 요소를 반환한다(take(1)과 비슷).
take(n)	데이터셋의 처음 n개 요소를 담은 배열을 반환한다.
takeSample(withReplacement, num, [seed])	데이터셋의 num개 요소를 무작위 샘플링한 배열을 반환한다. 복원 추출 또는 비복원 추출이며, 선택적으로 난수 생성기 시드를 미리 지정할 수 있다.
takeOrdered(n, [ordering])	자연 순서 또는 커스텀 비교기를 사용해 RDD의 처음 n개 요소를 반환한다.
saveAsTextFile(path)	데이터셋의 요소를 로컬 파일 시스템, HDFS 또는 Hadoop이 지원하는 다른 파일 시스템의 지정 디렉터리에 텍스트 파일(또는 텍스트 파일 집합)로 쓴다. Spark는 각 요소에 `toString`을 호출해 파일의 한 줄 텍스트로 변환한다.
saveAsSequenceFile(path)\n(Java 및 Scala)	데이터셋의 요소를 로컬 파일 시스템, HDFS 또는 Hadoop이 지원하는 다른 파일 시스템의 지정 경로에 Hadoop SequenceFile로 쓴다. Hadoop의 Writable 인터페이스를 구현하는 키-값 쌍 RDD에서 사용 가능하다. Scala에서는 Writable로 암묵 변환 가능한 타입에서도 쓸 수 있다(Spark는 Int, Double, String 등 기본 타입 변환을 포함한다).
saveAsObjectFile(path)\n(Java 및 Scala)	데이터셋의 요소를 Java 직렬화를 사용한 단순한 포맷으로 쓴다. 나중에 `SparkContext.objectFile()`으로 로드할 수 있다.
countByKey()	(K, V) 타입 RDD에서만 사용 가능하다. 각 키의 개수를 담은 (K, Int) 쌍 해시맵을 반환한다.
foreach(func)	데이터셋의 각 요소에 함수 func를 실행한다. 보통 Accumulator 갱신이나 외부 저장 시스템과의 상호작용 같은 부수 효과(side effects)를 위해 실행된다. 주의: Accumulator가 아닌 다른 변수를 `foreach()` 바깥에서 수정하면 정의되지 않은 동작이 생길 수 있다. 자세한 내용은 Understanding closures를 참고하세요.

Spark RDD API는 몇몇 액션의 비동기 버전도 제공해요. 예를 들어 foreach의 foreachAsync는 액션 완료를 기다리며 블로킹하는 대신 즉시 FutureAction을 호출자에게 반환해요. 이걸로 액션의 비동기 실행을 관리하거나 기다릴 수 있어요.

Shuffle operations

Spark 안의 어떤 연산은 셔플(shuffle) 이라는 이벤트를 촉발해요. 셔플은 데이터가 파티션들에 걸쳐 다르게 그룹화되도록 재분배하는 Spark의 메커니즘이에요. 이 과정은 보통 데이터를 실행기와 머신에 걸쳐 복사하기 때문에, 셔플은 복잡하고 비용이 큰 연산이에요.

Background

셔플 동안 무슨 일이 일어나는지 reduceByKey 연산을 예로 이해해 볼게요. reduceByKey는 각 키의 모든 값을 하나의 튜플로 결합한 새 RDD를 만들어요. 튜플은 키와 그 키에 연결된 모든 값에 reduce 함수를 실행한 결과로 이뤄져요. 문제는 한 키의 모든 값이 반드시 같은 파티션에(심지어 같은 머신에) 있지는 않지만, 결과를 계산하려면 함께 위치해야(co-located) 해요.

Spark에서 데이터는 특정 연산에 필요한 위치에 맞춰 파티션에 분산되지 않는 게 일반적이에요. 계산 중에 하나의 태스크는 하나의 파티션에서 동작해요. 그렇다면 단일 reduceByKey reduce 태스크를 실행하기 위해 모든 데이터를 모으려면, Spark는 all-to-all 연산을 수행해야 해요. 모든 키의 모든 값을 찾으려고 모든 파티션을 읽고, 각 키의 최종 결과를 계산하기 위해 파티션을 가로질러 값을 모아야 하지요. 이것이 바로 셔플이에요.

새로 셔플된 데이터의 각 파티션에 있는 요소 집합은 결정적(deterministic)이고 파티션 자체의 순서도 결정적이지만, 요소들의 순서는 결정적이지 않아요. 셔플 후 예측 가능하게 정렬된 데이터가 필요하다면 다음을 쓸 수 있어요.

mapPartitions로 각 파티션을 정렬(예: .sorted)
repartitionAndSortWithinPartitions로 재파티셔닝을 하면서 동시에 파티션을 효율적으로 정렬
sortBy로 전역적으로 정렬된 RDD 만들기

셔플을 일으킬 수 있는 연산으로는 repartition, coalesce 같은 재파티션 연산, groupByKey, reduceByKey 같은 'ByKey 연산(counting 제외), 그리고 cogroup, join 같은 조인 연산이 있어요.

Performance Impact

셔플은 디스크 I/O, 데이터 직렬화, 네트워크 I/O를 수반하므로 비용이 큰 연산이에요. 셔플용 데이터를 정리하기 위해 Spark는 태스크 집합을 만들어요. 데이터를 정리하는 map 태스크와 이를 집계하는 reduce 태스크 집합이죠. 이 명명은 MapReduce에서 온 것으로, Spark의 map, reduce 연산과 직접 관련되지는 않아요.

내부적으로 개별 map 태스크의 결과는 메모리에 들어갈 수 있을 때까지 유지돼요. 그러면 대상 파티션에 따라 정렬된 뒤 하나의 파일로 기록돼요. reduce 쪽에서는 관련 정렬된 블록을 읽니다.

일부 셔플 연산은 상당한 힙 메모리를 소비할 수 있어요. 레코드를 전송 전후에 정리할 때 인메모리 데이터 구조를 쓰기 때문이에요. 구체적으로 reduceByKey와 aggregateByKey는 map 쪽에 이런 구조를 만들고, 'ByKey 연산은 reduce 쪽에서 만든답니다. 데이터가 메모리에 안 들어가면 Spark는 이 테이블을 디스크로 넘겨(spill) 디스크 I/O와 가비지 컬렉션 증가라는 추가 비용이 생겨요.

셔플은 또한 디스크에 대량의 중간 파일을 만들어요. Spark 1.3부터 이 파일들은 해당 RDD가 더 이상 사용되지 않고 가비지 컬렉션될 때까지 보존돼요. 이는 lineage를 다시 계산해도 셔플 파일을 다시 만들 필요가 없게 하기 위함이에요. 애플리케이션이 이런 RDD에 대한 참조를 유지하거나 GC가 자주 발동하지 않으면, 가비지 컬렉션은 오랜 시간이 지난 뒤에야 일어날 수 있어요. 이는 오래 실행되는 Spark job이 대량의 디스크 공간을 소비할 수 있음을 의미해요. 임시 저장 디렉터리는 Spark 컨텍스트를 구성할 때 spark.local.dir 설정 파라미터로 지정해요.

셔플 동작은 다양한 설정 파라미터를 조정해 튜닝할 수 있어요. Spark Configuration Guide의 'Shuffle Behavior' 섹션을 참고하세요.

RDD Persistence

Spark의 가장 중요한 기능 중 하나는 데이터셋을 여러 연산에 걸쳐 **메모리에 영속화(또는 캐싱)**하는 거예요. RDD를 영속화하면 각 노드가 자신이 계산하는 파티션을 메모리에 저장하고, 그 데이터셋(또는 그에서 파생된 데이터셋)에 대한 다른 액션에서 재사용해요. 이 덕분에 이후의 액션은 훨씬 빨라질 수 있어요(종종 10배 이상). 캐싱은 반복 알고리즘과 빠른 대화형 사용의 핵심 도구예요.

RDD에 persist() 또는 cache() 메서드를 호출하면 영속화 대상으로 표시할 수 있어요. 액션에서 처음 계산되면 노드의 메모리에 보관돼요. Spark의 캐시는 장애 내성이 있어서, RDD의 어떤 파티션이 유실돼도 원래 그 파티션을 만든 트랜스포메이션으로 자동 재계산해요.

또한 각 영속화된 RDD는 **다른 저장 레벨(storage level)**로 저장할 수 있어요. 예를 들어 데이터셋을 디스크에 영속화하거나, 메모리에는 직렬화된 Java 객체로(공간 절약용) 영속화하거나, 노드에 걸쳐 복제할 수 있죠. 이런 레벨은 persist()에 StorageLevel 객체(Python, Scala, Java)를 넘겨 설정해요. cache() 메서드는 기본 저장 레벨, 즉 StorageLevel.MEMORY_ONLY(비직렬화 객체를 메모리에 저장)를 쓰는 약식이에요. 전체 저장 레벨 목록은 이렇습니다.

저장 레벨	의미
MEMORY_ONLY	RDD를 JVM에 비직렬화된 Java 객체로 저장한다. RDD가 메모리에 안 들어가면 일부 파티션은 캐시되지 않고 필요할 때마다 즉시 재계산된다. 기본 레벨이다.
MEMORY_AND_DISK	RDD를 JVM에 비직렬화된 Java 객체로 저장한다. 메모리에 안 들어가면 들어가지 못한 파티션을 디스크에 저장하고 필요할 때 거기서 읽는다.
MEMORY_ONLY_SER\n(Java 및 Scala)	RDD를 직렬화된 Java 객체로 저장한다(파티션당 하나의 바이트 배열). 빠른 직렬화기를 쓸 때 일반적으로 비직렬화 객체보다 공간 효율적이지만, 읽는 데 CPU를 더 쓴다.
MEMORY_AND_DISK_SER\n(Java 및 Scala)	MEMORY_ONLY_SER와 비슷하지만, 메모리에 안 들어가는 파티션을 매번 재계산하는 대신 디스크로 넘긴다.
DISK_ONLY	RDD 파티션을 디스크에만 저장한다.
MEMORY_ONLY_2, MEMORY_AND_DISK_2 등	위 레벨들과 같지만 각 파티션을 두 클러스터 노드에 복제한다.
OFF_HEAP (실험적)	MEMORY_ONLY_SER와 비슷하지만 데이터를 off-heap 메모리에 저장한다. off-heap 메모리가 활성화되어 있어야 한다.

참고: Python에서 저장된 객체는 항상 Pickle 라이브러리로 직렬화되므로 직렬화 레벨을 고르는 건 무의미해요. Python에서 쓸 수 있는 저장 레벨은 MEMORY_ONLY, MEMORY_ONLY_2, MEMORY_AND_DISK, MEMORY_AND_DISK_2, DISK_ONLY, DISK_ONLY_2, DISK_ONLY_3이에요.

Spark는 사용자가 persist를 호출하지 않아도 셔플 연산(예: reduceByKey)의 일부 중간 데이터를 자동으로 영속화하기도 해요. 셔플 중 노드가 실패할 때 전체 입력을 다시 계산하지 않게 하려는 거예요. 그래도 결과 RDD를 재사용할 계획이라면 persist를 호출하는 걸 권장해요.

Which Storage Level to Choose?

Spark의 저장 레벨은 메모리 사용량과 CPU 효율 사이의 서로 다른 트레이드오프를 제공하려는 거예요. 선택할 때는 다음 과정을 따라가는 걸 권장해요.

RDD가 기본 저장 레벨(MEMORY_ONLY)에 부담 없이 들어간다면 그대로 두세요. 이것이 가장 CPU 효율적인 옵션이라 RDD 연산을 최대한 빠르게 실행할 수 있어요.
안 들어간다면 MEMORY_ONLY_SER을 시도하고, 빠른 직렬화 라이브러리를 골라 객체를 훨씬 공간 효율적으로 만들되 여전히 접근은 합리적으로 빠르게 유지하세요. (Java 및 Scala)
데이터셋을 계산한 함수가 비싸거나 데이터를 대량으로 필터링하지 않는 한 디스크로 넘기지(spill) 마세요. 그렇지 않으면 디스크에서 읽는 것만큼 파티션을 재계산하는 게 빠를 수 있어요.
빠른 장애 복구가 필요하다면(예: Spark로 웹 애플리케이션 요청을 서빙하는 경우) 복제 저장 레벨을 쓰세요. 모든 저장 레벨은 유실 데이터를 재계산해 완전한 장애 내성을 제공하지만, 복제된 레벨은 유실 파티션 재계산을 기다리지 않고 RDD에서 태스크를 계속 실행하게 해줘요.

Removing Data

Spark는 각 노드의 캐시 사용량을 자동으로 모니터링하고, 가장 오래 사용되지 않은(LRU, least-recently-used) 방식으로 오래된 데이터 파티션을 버려요. 캐시에서 빠질 때까지 기다리는 대신 RDD를 수동으로 제거하고 싶다면 RDD.unpersist() 메서드를 쓰세요. 이 메서드는 기본적으로 블로킹하지 않아요. 리소스가 해제될 때까지 블로킹하려면 호출할 때 blocking=true를 지정하면 됩니다.

Shared Variables

보통 Spark 연산(예: map, reduce)에 넘긴 함수는 원격 클러스터 노드에서 실행될 때, 함수에 쓰인 모든 변수의 별도 복사본에서 동작해요. 이 변수들은 각 머신으로 복사되고, 원격 머신에서 변수에 대한 갱신은 드라이버 프로그램으로 되돌려지지 않아요. 태스크들 사이에서 일반적이고 읽기-쓰기 가능한 공유 변수를 지원하는 건 비효율적이에요. 하지만 Spark는 두 가지 흔한 사용 패턴을 위해 두 종류의 제한된 공유 변수를 제공해요. 바로 브로드캐스트 변수와 어큐뮬레이터입니다.

Broadcast Variables

브로드캐스트 변수는 개발자가 읽기 전용 변수를 각 머신에 캐시해서, 태스크마다 복사본을 보내지 않게 해줘요. 예를 들어 큰 입력 데이터셋의 복사본을 모든 노드에 효율적으로 줄 때 쓸 수 있어요. Spark는 또한 효율적인 브로드캐스트 알고리즘으로 브로드캐스트 변수를 분배해 통신 비용을 줄이려고 해요.

Spark 액션은 일련의 스테이지(stage)로 실행되는데, 각 스테이지는 분산 "셔플" 연산으로 구분돼요. Spark는 각 스테이지 안의 태스크들이 필요로 하는 공통 데이터를 자동으로 브로드캐스트해요. 이렇게 브로드캐스트된 데이터는 직렬화된 형태로 캐시되고 각 태스크 실행 전에 역직렬화돼요. 즉, 여러 스테이지의 태스크가 같은 데이터를 필요로 하거나, 데이터를 비직렬화된 형태로 캐시하는 게 중요할 때만 브로드캐스트 변수를 명시적으로 만드는 게 유용해요.

브로드캐스트 변수는 변수 v에 SparkContext.broadcast(v)를 호출해 만들어요. 브로드캐스트 변수는 v의 래퍼이고, 그 값은 value 메서드를 호출해 접근할 수 있어요. 아래 코드가 이를 보여줍니다.

>>> broadcastVar = sc.broadcast([1, 2, 3])
<pyspark.core.broadcast.Broadcast object at 0x102789f10>

>>> broadcastVar.value
[1, 2, 3]

브로드캐스트 변수를 만든 뒤에는 클러스터에서 실행되는 어떤 함수에서도 값 v 대신 이 변수를 써야 해요. 그래야 v가 노드로 여러 번 보내지지 않으니까요. 또한 v 객체는 브로드캐스트한 뒤 수정하면 안 돼요. 그래야 모든 노드가 브로드캐스트 변수의 같은 값을 받을 수 있으니까요(예: 나중에 새 노드로 변수가 보내질 때).

실행기에 복사된 리소스를 해제하려면 .unpersist()를 호출해요. 그 후 브로드캐스트를 다시 쓰면 다시 브로드캐스트돼요. 브로드캐스트 변수가 쓰는 모든 리소스를 영구히 해제하려면 .destroy()를 호출해요. 그다음부터는 브로드캐스트 변수를 쓸 수 없어요. 이 메서드들은 기본적으로 블로킹하지 않는다는 점을 참고하세요. 리소스가 해제될 때까지 블로킹하려면 호출할 때 blocking=true를 지정하면 됩니다.

Accumulators

어큐뮬레이터는 결합·교환 법칙을 만족하는 연산으로 "더하기만" 할 수 있는 변수라서 병렬로 효율적으로 지원할 수 있어요. MapReduce의 카운터나 합계를 구현할 때 쓸 수 있죠. Spark는 숫자 타입의 어큐뮬레이터를 기본 지원하고, 개발자가 새 타입에 대한 지원을 추가할 수도 있어요.

사용자는 이름 있는(name) 또는 이름 없는(unnamed) 어큐뮬레이터를 만들 수 있어요. 아래 이미지처럼 이름 있는 어큐뮬레이터(여기선 counter)는 그 어큐뮬레이터를 수정한 스테이지의 웹 UI에 표시돼요. Spark는 태스크가 수정한 각 어큐뮬레이터의 값을 "Tasks" 테이블에 보여줍니다.

UI에서 어큐뮬레이터를 추적하면 실행 중인 스테이지의 진행 상황을 이해하는 데 유용할 수 있어요(참고: Python에서는 아직 지원되지 않아요).

어큐뮬레이터는 초기 값 v에 SparkContext.accumulator(v)를 호출해서 만들어요. 클러스터에서 실행되는 태스크는 add 메서드나 += 연산자로 값을 더할 수 있지만, 값을 읽을 수는 없어요. 값은 드라이버 프로그램만 value 메서드로 읽을 수 있답니다.

아래 코드는 어큐뮬레이터로 배열의 요소를 더하는 모습을 보여줘요.

>>> accum = sc.accumulator(0)
>>> accum
Accumulator<id=0, value=0>

>>> sc.parallelize([1, 2, 3, 4]).foreach(lambda x: accum.add(x))
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

>>> accum.value
10

이 코드는 Int 타입 어큐뮬레이터의 내장 지원을 썼지만, 개발자는 AccumulatorParam을 서브클래싱해 자신만의 타입을 만들 수도 있어요. AccumulatorParam 인터페이스에는 메서드 두 개가 있어요. 데이터 타입의 "zero 값"을 제공하는 zero와, 두 값을 더하는 addInPlace죠. 예를 들어 수학적 벡터를 나타내는 Vector 클래스가 있다면 이렇게 쓸 수 있어요.

class VectorAccumulatorParam(AccumulatorParam):
    def zero(self, initialValue):
        return Vector.zeros(initialValue.size)

    def addInPlace(self, v1, v2):
        v1 += v2
        return v1

# Then, create an Accumulator of this type:
vecAccum = sc.accumulator(Vector(...), VectorAccumulatorParam())

액션 안에서만 수행된 어큐뮬레이터 갱신에 대해 Spark는 각 태스크의 갱신이 정확히 한 번만 적용되도록 보장해요. 즉 재시작된 태스크는 값을 갱신하지 않아요. 트랜스포메이션에서는 태스크나 job 스테이지가 재실행되면 각 태스크의 갱신이 두 번 이상 적용될 수 있다는 점을 알아야 해요.

어큐뮬레이터는 Spark의 지연 평가 모델을 바꾸지 않아요. RDD에 대한 연산 안에서 갱신되더라도, 그 RDD가 액션의 일부로 계산되어야만 값이 갱신돼요. 그 결과 어큐뮬레이터 갱신은 map() 같은 lazy 트랜스포메이션 안에서 이루어지면 실행이 보장되지 않아요. 아래 코드 조각이 이 성질을 보여줍니다.

accum = sc.accumulator(0)
def g(x):
    accum.add(x)
    return f(x)
data.map(g)
# Here, accum is still 0 because no actions have caused the `map` to be computed.

Deploying to a Cluster

애플리케이션 제출 가이드에서 애플리케이션을 클러스터에 제출하는 방법을 설명해요. 요약하면, 애플리케이션을 JAR(Java/Scala) 또는 .py/.zip 파일 묶음(Python)으로 패키징하면, bin/spark-submit 스크립트가 지원되는 어떤 클러스터 매니저로든 제출할 수 있게 해줘요.

Launching Spark jobs from Java / Scala

org.apache.spark.launcher 패키지는 간단한 Java API로 Spark job을 자식 프로세스로 실행하는 클래스를 제공해요.

Unit Testing

Spark는 어떤 인기 있는 유닛 테스트 프레임워크와도 잘 어울려요. 테스트에서 master URL을 local로 설정한 SparkContext를 만들고, 연산을 실행한 뒤, SparkContext.stop()을 호출해 정리하면 됩니다. 같은 프로그램에서 컨텍스트 두 개가 동시에 실행되는 건 Spark가 지원하지 않으므로, finally 블록이나 테스트 프레임워크의 tearDown 메서드 안에서 컨텍스트를 반드시 멈추도록 하세요.

Where to Go from Here

Spark 웹사이트에서 몇 가지 예제 Spark 프로그램을 볼 수 있어요. 또한 Spark는 examples 디렉터리에 여러 샘플(Python, Scala, Java, R)을 포함해요. Java와 Scala 예제는 클래스 이름을 Spark의 bin/run-example 스크립트에 넘겨 실행할 수 있어요. 예를 들면 이렇게요.

./bin/run-example SparkPi

Python 예제는 spark-submit을 대신 써요.

./bin/spark-submit examples/src/main/python/pi.py

R 예제도 spark-submit을 써요.

./bin/spark-submit examples/src/main/r/dataframe.R

프로그램 최적화에 대한 도움은 설정 및 튜닝 가이드가 모범 사례를 알려줘요. 특히 데이터를 효율적인 포맷으로 메모리에 저장하는 데 중요하죠. 배포에 대한 도움은 클러스터 모드 개요가 분산 운영과 지원되는 클러스터 매니저의 구성 요소를 설명해줘요.

마지막으로, 전체 API 문서는 Python, Scala, Java, R에서 제공돼요.