Spark (4) - Spark의 데이터 종류 및 처리법 - RDD 편

Spark의 데이터 모델에 대해 학습하고,

각 모델에서는 어떤 API를 활용할 수 있는지 학습해보자.

---

1. Spark Data Models

Data Model(데이터 모델)?
복잡한 물리적 데이터(디스크에 저장된 바이트 단위의 조각들)를 개발자가 코드 수준에서 어떻게 바라볼지 결정하는 추상화 단계

Spark에서의 데이터 모델
수만 대의 서버에 흩어진 데이터 조각들을 "하나의 리스트(RDD)"로 볼 것인지,
아니면 "하나의 표(DataFrame)"로 볼 것인지를 결정하는 규칙이다.

Spark의 대표적인 DataModel 세 가지

• RDD: 가장 로우 레벨의 데이터에 사용함
• DataFrame: 스키마 있는 데이터에 사용함
• Dataset: 데이터의 타입이 정해져 있는 정적인 타입인 경우 사용함

---

2. RDD

2-1. RDD 란?

Resilient(탄력있는, 변함없는, 회복력있는) Distributed Dataset

회복력 있는?
=> 데이터가 유실되어도 복구하기 쉬움,
=> 한번 생성된 후 변경할 수 없음

• 분산 데이터 모델
• MapReduce와 비슷한 역할
  • Map: 데이터를 쪼개고 변환한다. (예: 문장을 단어로 쪼개기)
  • Reduce: 쪼개진 것들을 다시 합쳐서 결과를 낸다. (예: 단어 개수 합치기)
• How(어떤 작업을 할지)를 기술
  • "첫 번째 줄을 가져와서 공백으로 자르고, 튜플로 만든 다음에, 키별로 더해줘"라고 단계별 방법(How)을 일일이 명령
  • 뒤에서 배울 DataFrame은 반대로 What(무엇을 원하는지)만 말한다.
• 기본적으로 여러 클러스터의 메모리 상에 데이터가 올라가게 된다.
  • 데이터를 하드디스크(느림)에 저장하지 않고 메모리(RAM, 빠름)에 올려두고 처리한다.
• Low-Level 데이터 모델
  • 데이터의 Schema를 Spark가 모르고 Spark 입장에서는 그저 "정체를 알 수 없는 객체들의 묶음"일 뿐이다.

2-2. RDD 특성

• resilient (불변성), Read-only
• Fault-tolerance, 복구 가능 (RDD Lineage-RDD가 DAG 형태로 저장되는 것)
  • 데이터를 복제해 두는 게 아니라, "어떻게 만들었는지(DAG)"를 기록해 두기 때문
• MR(Map Reduce)과 유사한 방식이지만 memory 적재하기 때문에 좀 더 빠르다
  • MapReduce는 중간 결과를 매번 하드디스크에 썼다 읽었다 해서 느린데, Spark는 최대한 RAM에서 처리한다.
• Lazy Evaluation (지연 실행)
  • 코드를 썼다고 바로 실행되지 않는다.

2-3. RDD의 연산 1. Transformation

RDD 를 새로운 RDD 로 변형 (read-only 특성)
RDD 는 Read-only라는 특성땜에 RDD를 변경해서 바로 쓰는게 아니라
RDD 변경 -> 새로운 RDD로 변형해서 변형된 RDD를 사용하는거다.

 

RDD Lineage

RDD 가 만들어진 일종의 족보 개념이다
A->B->C 순서로 가공되었다면 C가 유실되어도 Lineage를 보고 A부터 다시 만들면 된다.

• transformation 이 적용되는 연산 순서를 나타내는 DAG
• memory 에서 유실 되더라도 DAG 를 따라서 데이터 복구 가능

Transformation 함수만 실행하면 결과값(데이터)이 나오지 않는다.

그냥 "이런 작업을 할 거야"라고 기록만 된다. (Lazy-Evaluation)

 

Transformation API

• map(): 각 row에 함수를 적용하고 결과 RDD 반환
• filter(): 조건을 만족하는 row만 남겨서 반환
• union(): 두 RDD의 합집합 RDD 반환
• distinct(): RDD 내의 중복값을 제거한 RDD 반환 

2-4. RDD의 연산 2. Action

Spark의 핵심철학인 Lazy Evaluation 때문에
map, filter 와 같은 Transformation 만 써서는 아무일도 일어나지 않는다.

• 실제 연산을 발생시키는 명령
• 리턴 값이 데이터 이거나 실행 결과
• Action 에 해당하는 API 가 실행되기 전까지 실제 연산은 수행되지 않음

 

• 원리: Action API가 호출되는 순간, Spark는 지금까지 쌓인 "계보(Lineage)"를 쭉 훑어본다.
  "아, 사용자가 드디어 결과를 보겠다고 하네? 그럼 가장 효율적인 경로(DAG)를 짜서 실행하자!"라며 서버들을 가동하는것
• 리턴 값: Transformation이 새로운 RDD(또 다른 설계도)를 돌려줬다면,
  Action은 실제 데이터(List, Integer 등)나 성공 여부를 돌려줌

자세한 API 들은 아래 문서에서 자세히 확인할 수 있다.

https://spark.apache.org/docs/latest/api/python/reference/pyspark.html

Spark Core — PySpark 4.1.1 documentation

 

Action API

• collect(): RDD의 모든 값 리턴
• count(): RDD의 값 개수 연산 리턴
• reduce(f): 병렬로 f 함수에 해당하는 병합 연산 리턴
• take(n): 상위 n 만큼의 개수를 리턴

---

3. RDD 만드는 법

1) local 파이썬 데이터셋을 RDD 데이터 셋으로 변환 하는 방법

data = ["one", "two", "three"]
sc.parallelize(data)

 

2) range 함수 사용

sc.range(1,6)

 

3) read file

sc.textFile("data/data1.txt")

 

 

예시)

rdd = sc.parallelize([1, 2, 3, 4, 5])
rdd.filter(lambda x: x % 2 == 0).collect()

spark context 의 parallelize 를 통해 [1,2,3,4,5]를 RDD 값으로 변경하였고

이 값 중에서 짝수인 값만 따로 filter 를 통해서 선별해서

collect()를 통해 조건에 만족하는 모든 RDD 값을 리턴하도록 한 코드이다.

---

4. 추가적인 실습

4-1. RDD의 파티션 처리 glom()

rdd = sc.range(0, 100, 1, 5)
rdd.getNumPartitions() # 결과는 5가 나옴

위 코드를 실핼시키면 5가 나오는데, 

이는 range를 통해서 생성된 100개의 숫자가 5개의 통(파티션)에 20개씩 나누어져 들어갔을 거란 뜻이다.

 

아래 코드를 보자.

rdd = sc.range(0, 100, 1, 5)
rdd1 = rdd.glom().map(lambda arr: len(arr))
rdd1.take(10)

# 실행 결과
[20, 20, 20, 20, 20]

 

rdd.glom()은

각 파티션 안에 흩어져 있는 요소들을 하나의 리스트로 묶는다.

이해하기 쉽게 아래와 같이 비유해봤다.

• 기존: [0, 1, 2, ...] (100개의 개별 아이템)
• glom 후: [[0, 1, ...19], [20, ...39], [40, ...59], [60, ...79], [80, ...99]]

즉, 파티션마다 아이템들을 '박싱(Boxing)' 처리해서 5개의 커다란 배열로 만드는 작업이다.