[Spark] RDD와 데이터프레임

개요

스파크에서는 RDD(Resilient Distributed Dataset)와 데이터프레임 두 가지 방법으로 데이터를 처리하는데 각가 다른 추상화 수준과 특징을 가지고 있습니다.

RDD(Resilient Distributed Dataset)

스파크의 가장 기본적이고 낮은 수준의 데이터 추상화입니다. 여러 서버에 분산되어 병렬로 처리될 수 있는 변경 불가능한(immutable) 요소들의 집합입니다. (이 부분은 데이터프레임도 가지고 있는 특징)

RDD는 다음과 같은 장단점을 가지고 있습니다.

장점

• RDD는 내부에 어떤 타입의 객체(Python, Java, Scala)든 담을 수 있으며 데이터의 스키마에 대한 정보를 가지지 않습니다. 즉, 스키마가 없는 비정형 데이터(예: 일반 텍스트 로그 파일, 바이너리 데이터)를 유연하게 다룰 수 있습니다.
• 저수준 API를 통해 개발자가 직접 데이터 처리 방식(예: 파티셔닝, 스토리지 등)을 세밀하게 제어할 수 있습니다.
• Scala, Java의 경우 컴파일 시점에 타입을 체크하여 오류를 잡을 수 있습니다

단점

• Catalyst와 같은 자동 최적화 엔진이 없어 개발자가 직접 최적화하지 않는 한 데이터프레임에 비해 성능이 떨어집니다.
• map, flatMap, reduceByKey 등 저수준 API를 사용하므로 코드가 더 길고 복잡해집니다.

다음은 100GB 비정형 로그 파일에서 ‘ERROR’라는 단어가 포함된 라인만 카운트하는 예시입니다.

# 텍스트 파일을 RDD로 로드
log_rdd = spark.sparkContext.textFile("server_log.txt")

# 'ERROR'가 포함된 라인만 필터링 (Transformation)
error_rdd = log_rdd.filter(lambda line: "ERROR" in line)

# 필터링된 라인의 수를 계산 (Action)
error_count = error_rdd.count()

print(f"Total error lines: {error_count}")

• log_rdd는 로그 파일의 각 줄(line)을 하나의 요소로 갖는 거대한 컬렉션입니다.
• 스파크는 100GB 파일을 하나의 서버로 가져오지 않습니다. 파일을 여러 조각(파티션)으로 나누어 10대의 서버(워커 노드)에 각각 10GB씩 분산하여 저장합니다.
• filter 연산이 실행될 때, 10대의 서버는 각자 자신이 맡은 10GB의 데이터에 대해 동시에 'ERROR'라는 단어를 검색합니다. 작업이 병렬로 처리되므로 매우 빠르게 끝납니다.
• log_rdd.filter(...) 연산은 원본 log_rdd를 수정하지 않습니다. 필터링 조건에 맞는 요소들로 구성된 새로운 RDD(error_rdd)를 생성하여 반환합니다. 원본은 그대로 유지되기 때문에 작업 도중 일부 서버에 문제가 생겨도 스파크는 원본 RDD로부터 다시 계산하여 작업을 복구할 수 있습니다. 이 부분을 Resilient라고 부릅니다. (Immutable)

데이터프레임

RDD 위에 구축된 더 높은 수준의 추상화입니다. 데이터를 이름이 있는 열(column)으로 구성된 테이블 형태로 구조화하여 다룹니다. Pandas나 R의 데이터프레임과 유사합니다.

데이터프레임은 다음과 같은 장단점을 가지고 있습니다.

장점

• 모든 데이터에 스키마가 강제되어 명확한 열 이름과 타입을 가지므로 데이터의 정합성 유지가 쉽고 오류를 줄일 수 있습니다.
• Catalyst라는 쿼리 최적화 엔진을 통해 개발자가 작성한 코드를 스파크가 분석하여 가장 효율적인 실행 계획을 세웁니다. 이로 인해 대부분의 경우 RDD보다 훨씬 뛰어난 성능을 보입니다.
• SQL과 유사한 고수준 API(select, filter, groupBy 등)를 제공하여 데이터를 쉽게 조작할 수 있습니다.
• JSON, CSV, Parquet, JDBC 등 수많은 데이터 소스를 쉽게 읽고 쓸 수 있습니다.

단점

• 최적화가 자동으로 이루어지므로 파티셔닝 등 저수준의 물리적 데이터 제어가 어렵습니다.
• Python의 경우, 실행 시점까지 열의 이름이나 타입 오류를 발견하기 어렵습니다.

다음은 RDD 동일하게 100GB 비정형 로그 파일에서 ‘ERROR’라는 단어가 포함된 라인만 카운트하는 예시입니다.

# 텍스트 파일을 데이터프레임으로 로드
# 각 줄은 'value'라는 기본 열(column) 이름으로 저장됩니다.
log_df = spark.read.text("server_log.txt")

# 'value' 열의 내용에 'ERROR'가 포함된 행(row)만 필터링
error_df = log_df.filter(log_df.value.contains("ERROR"))

# 필터링된 행의 수를 계산
error_count = error_df.count()

print(f"Total error lines: {error_count}")

## 스파크 SQL 구현
# 텍스트 파일을 데이터프레임으로 로드
log_df = spark.read.text("server_log.txt")

# 'logs'라는 이름의 임시 뷰로 등록
log_df.createOrReplaceTempView("logs")

# Spark SQL을 사용하여 동일한 작업 수행
error_df = spark.sql("SELECT * FROM logs WHERE value LIKE '%ERROR%'")

# 필터링된 행의 수를 계산
error_count = error_df.count()

print(f"Total error lines: {error_count}")

데이터프레임 API(filter(), select() 등)와 스파크SQL은 내부적으로 동일한 Catalyst 옵티마이저를 통해 처리됩니다. 두 방식 모두 최종적으로 동일한 물리적 실행 계획으로 변환되어 실행되므로 성능 차이는 거의 없습니다. 선택은 개발자의 선호도나 프로젝트의 코드 스타일에 따라 달라집니다.

RDD를 반드시 사용해야 하는 경우

데이터프레임이 훨씬 우수하지만, 아래와 같은 특정 상황에서는 RDD를 사용해야만 합니다.

1. 데이터가 테이블 형태(행과 열)로 표현할 수 없는 경우 (예: 유전체 데이터, 과학 계산 데이터, 그래프 데이터 등)
2. 스파크의 파티셔닝 방식 등을 매우 세밀하게 직접 제어하여 특정 성능 최적화를 해야만 하는 고급 사용 사례의 경우(=저수준의 물리적 제어가 필요할 때)
3. Spark 초기에 만들어진 일부 머신러닝이나 그래프 처리 라이브러리(MLlib, GraphX의 일부)는 RDD를 기본 데이터 타입으로 요구하는 경우가 있음

결론

Catalyst 최적화 엔진과 사용의 편의성 때문에 특별한 이유가 없다면 데이터프레임을 사용하는 것이 좋습니다.