Spark RDD 프로그래밍 가이드

Spark RDD 프로그래밍 가이드 (RDD Programming Guide)

RDD는 Spark의 가장 오래된 추상화이자 분산 데이터 처리의 근간을 이루는 개념이에요. 지금은 Dataset/DataFrame이 주력이지만, 파티션이나 트랜스포메이션 같은 분산 처리의 동작 방식을 이해하려면 RDD를 먼저 보는 게 가장 확실합니다. 이 페이지에서는 RDD가 무엇인지, 어떻게 만들고 인메모리에 보존하는지부터 차근차근 설명할게요.

출처: Apache Spark 공식 문서 – RDD Programming Guide

개요 (Overview)

모든 Spark 애플리케이션은 사용자의 main 함수를 실행하는 드라이버 프로그램과, 클러스터에서 다양한 병렬 연산을 수행하는 구조로 이뤄져 있어요. Spark가 제공하는 핵심 추상화는 RDD(Resilient Distributed Dataset, 회복성 분산 데이터셋) 인데, 클러스터의 노드들에 분할(파티션)되어 병렬로 연산할 수 있는 요소들의 모음입니다.

RDD는 Hadoop 파일 시스템(또는 Hadoop이 지원하는 다른 파일 시스템)의 파일, 또는 드라이버 프로그램의 기존 Scala 컬렉션에서 시작해 변환해서 만듭니다. 그리고 Spark에 RDD를 인메모리에 persist(보존)하도록 요청해 병렬 연산 사이에 효율적으로 재사용할 수도 있어요. 무엇보다 RDD는 노드 장애가 나면 자동으로 복구됩니다.

Spark의 두 번째 추상화는 공유 변수(shared variables) 입니다. 기본적으로 Spark가 병렬 연산을 노드별 task로 나눠 실행할 때, 함수에 쓰인 각 변수의 사본을 task마다 배달해요. 그런데 어떤 변수는 task끼리, 또는 task와 드라이버 프로그램 사이에 공유해야 할 때가 있어요. Spark는 두 종류의 공유 변수를 지원합니다.

  • 브로드캐스트 변수(broadcast variables): 모든 노드의 메모리에 값을 캐시해 재사용
  • 어큐뮬레이터(accumulators): 카운터나 합계처럼 "더하기만" 되는 변수

이 가이드는 각 언어에서 이 기능들을 어떻게 쓰는지 보여주는데, Spark의 인터랙티브 셸(bin/pyspark 또는 bin/spark-shell)에서 따라 해보면 가장 쉽게 익힐 수 있어요.

Spark 초기화 (Initializing Spark)

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

from pyspark import SparkContext, SparkConf

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

appName은 클러스터 UI에 표시될 애플리케이션 이름이고, master는 Spark/YARN 클러스터 URL 또는 로컬 모드로 돌리기 위한 특별한 "local" 문자열이에요. 실제 클러스터에서 운영할 땐 master를 프로그램에 하드코딩하지 말고 spark-submit으로 애플리케이션을 제출해서 받아오는 게 맞아요. 다만 로컬 테스트나 유닛 테스트에서는 "local"을 넘겨 프로세스 안에서 Spark를 실행할 수 있습니다.

PySpark 셸에서는 이미 인터프리터 인식용 SparkContext가 sc 변수로 만들어져 있어서, 직접 만드는 게 오히려 안 돼요. --master 인자로 연결 대상을 정하고, --py-files로 Python 파일을, --packages로 Maven 좌표 기반 의존성을 추가할 수 있어요.

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

IPython이나 Jupyter 노트북에서도 쓸 수 있어요. PYSPARK_DRIVER_PYTHON=ipython 또는 PYSPARK_DRIVER_PYTHON=jupyter PYSPARK_DRIVER_PYTHON_OPTS=notebook 환경변수를 지정하면 됩니다.

RDD 만들기

RDD를 만드는 방법은 두 가지예요. 드라이버 프로그램의 기존 컬렉션을 병렬화(parallelize) 하거나, 공유 파일시스템·HDFS·HBase·Hadoop InputFormat을 제공하는 데이터 소스의 데이터셋을 참조하는 겁니다.

병렬화된 컬렉션 (Parallelized Collections)

드라이버 프로그램의 기존 iterable/컬렉션에 SparkContext의 parallelize 메서드를 호출하면, 컬렉션의 요소가 복사되어 병렬 연산이 가능한 분산 데이터셋이 돼요.

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

만든 뒤엔 distData.reduce(lambda a, b: a + b)처럼 병렬로 연산할 수 있어요. 중요한 파라미터가 하나 있는데, 데이터셋을 몇 조각(파티션)으로 자를지예요. Spark는 파티션마다 하나의 task를 실행하므로 보통 클러스터 CPU 하나당 2~4개 파티션을 권장합니다. 대개 Spark가 자동으로 잡아주지만, sc.parallelize(data, 10)처럼 두 번째 인자로 직접 지정할 수도 있어요. (코드 일부에서 "slices"라는 용어는 파티션의 동의어로, 하위 호환을 위해 남아 있어요.)

외부 데이터셋 (External Datasets)

PySpark는 로컬 파일시스템, HDFS, Cassandra, HBase, Amazon S3 등 Hadoop이 지원하는 어떤 저장소에서든 분산 데이터셋을 만들 수 있어요. 텍스트 파일·SequenceFile·기타 Hadoop InputFormat을 지원합니다.

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

파일을 읽을 때 몇 가지 주의할 점이 있어요.

  • 로컬 파일시스템 경로를 쓸 땐 워커 노드에서도 같은 경로로 접근 가능해야 합니다. 모든 워커에 파일을 복사하거나 네트워크 마운트된 공유 파일시스템을 써야 해요.
  • textFile을 비롯한 파일 기반 입력 메서드는 디렉터리, 압축 파일, 와일드카드도 지원해요. 예: textFile("/my/directory"), textFile("/my/directory/*.txt"), textFile("/my/directory/*.gz").
  • textFile은 파일의 파티션 수를 제어하는 선택적 두 번째 인자도 받아요. 기본적으로 파일 블록(HDFS 기본 128MB)마다 파티션 하나를 만들지만, 더 큰 값을 넘기면 더 많은 파티션을 요청할 수 있어요. 단, 파티션이 블록보다 적어질 수는 없습니다.

텍스트 파일 외에도 Python API는 여러 포맷을 지원해요.

  • SparkContext.wholeTextFiles : 여러 작은 텍스트 파일이 든 디렉터리를 읽어 (파일명, 내용) 쌍으로 반환 (텍스트 파일의 textFile은 파일당 줄 단위 레코드를 반환)
  • RDD.saveAsPickleFile / SparkContext.pickleFile : pickle된 Python 객체로 RDD를 저장/로드 (기본 배치 크기 10)
  • SequenceFile 및 Hadoop Input/Output Format : 이 기능은 현재 Experimental로 표시돼 있으며 고급 사용자 대상이에요.

RDD 연산

RDD는 두 종류의 연산을 지원합니다.

  • 트랜스포메이션(transformations): 기존 RDD에서 새 RDD를 만드는 지연(lazy) 연산. 예: map, filter, flatMap
  • 액션(actions): 결과를 드라이버에 반환하거나 외부 저장소에 쓰는 연산. 예: reduce, collect, count

트랜스포메이션의 자세한 목록과 각 동작, 그리고 키-값 쌍 RDD를 다루는 reduceByKey 같은 연산은 본 위키의 "RDD 변환 (Transformations)" 페이지를 참고하세요.

RDD 보존 (Persistence)

Spark의 가장 중요한 기능 중 하나가 연산 사이에 RDD를 메모리에 persist 하는 거예요. RDD를 persist하면 각 노드는 셔플 연산 후 자신이 가진 파티션을 메모리에 저장하고, 이후 액션에서 재사용합니다. 덕분에 반복 알고리즘이나 인터랙티브 탐색에서 성능이 크게 빨라져요.

rdd = sc.parallelize(range(1, 10))
rdd.persist()  # 또는 rdd.cache()

cache()는 persist()를 기본 저장 수준(MEMORY_ONLY)으로 호출하는 약어예요. 메모리에 맞지 않으면 파티션이 재계산됩니다. MEMORY_AND_DISK처럼 저장 수준은 StorageLevel로 다양하게 지정할 수 있어요. 저장할 값이 충분히 큰지, 모든 값이 메모리에 들어갈지에 따라 저장 수준을 고르면 됩니다. (자세한 저장 수준과 메모리 튜닝은 본 위키의 "튜닝 (Tuning)" 페이지 참고.) persist된 RDD는 unpersist()로 해제할 수 있어요.

공유 변수 (Shared Variables)

기본적으로 Spark는 각 task에 함수에 쓰인 변수의 사본을 배달해요. 하지만 변수를 task들 사이에서 공유해야 할 때가 있는데, Spark는 두 가지 공유 변수를 제공합니다.

  • 브로드캐스트 변수: 큰 읽기 전용 값을 모든 노드에 효율적으로 배포. 각 task가 사본을 만들지 않아 통신 비용을 줄여줘요. sc.broadcast(v)로 생성하고 .value로 읽습니다.
  • 어큐뮬레이터: 카운터나 합계처럼 "더하기만" 되는 변수. 병렬 연산에서 집계하는 데 쓰이며, 드라이버만 .value를 읽을 수 있어요. sc.accumulator(0)로 생성합니다.

자세한 내용은 본 위키의 "공유 변수 (Shared Variables)" 페이지를 참고하세요.

더 알아보기