Spark 작업을 실행하면 실제로 어떤 일이 일어날까?

소개

데이터 세트가 커지면 CPU 용량, 메모리, 저장 장치 처리량의 한계로 인해 한 대의 시스템에서 데이터를 효율적으로 처리하기가 어려워질 수 있습니다.

Apache Spark는 작업을 파티션으로 나누고 CPU 코어와 여러 시스템에서 작업을 병렬로 실행할 수 있는 분산 데이터 처리 엔진을 제공해 이러한 문제를 해결합니다.

하지만 Spark는 단순히 시스템을 더 추가하는 것만을 의미하지 않습니다.

드라이버, 실행기, RDD, DataFrame, 파티션, 작업, 스테이지, 태스크 등 각 구성 요소가 어떻게 함께 작동하는지 이해하는 것은 효율적이고 안정적인 데이터 파이프라인을 설계하는 데 필수적입니다.

이 글에서는 전자상거래 매출 집계 사례를 통해 Spark의 아키텍처와 실행 방식, 주요 성능 고려 사항을 살펴보겠습니다.

1. Apache Spark를 사용하는 이유 – 병렬 데이터 처리

Apache Spark는 대규모 데이터 처리와 데이터 엔지니어링을 위해 설계된 분산 컴퓨팅 엔진입니다.

Spark는 병렬 데이터 처리를 활용해 빅데이터를 처리합니다:

그림 A. Apache Spark 분산 클러스터 아키텍처 개요(클러스터, 드라이버 및 워커 노드). 단일 노드 실행과 비교 (Kuriko IWAI 제작)

단일 노드 처리(오른쪽, 그림 A)와 달리 Spark는 여러 노드를 사용해 작업을 태스크로 나누고, 사용 가능한 경우 여러 시스템의 리소스에 걸쳐 태스크를 실행합니다.

여기서 분산이란 여러 시스템이 반드시 필요하다는 뜻은 아닙니다. Spark는 단일 시스템에서 로컬 모드로 실행할 수 있습니다. 이 경우에도 데이터를 파티션으로 나누어 병렬 처리할 수 있습니다.

2. Apache Spark 핵심 아키텍처

Spark는 다음과 같은 주요 구성 요소로 이루어져 있습니다:

• Spark 드라이버(드라이버): Spark 애플리케이션의 중앙 조정 프로세스입니다.
• 클러스터 관리자: CPU 코어와 메모리 같은 하드웨어 리소스를 할당하는 외부 서비스입니다.
• 워커 노드(워커): 클러스터를 구성하는 물리적 또는 가상 머신이나 컴퓨팅 노드입니다.
• 실행기: 워커 노드에서 실행되는 분산 워커 프로세스입니다.
• Java Virtual Machine (JVM): Spark의 바이트코드를 실행하는 데 필요한 런타임 환경입니다(Spark는 Scala/Java로 작성되었기 때문입니다).

먼저 드라이버는 Spark 애플리케이션을 조정하고 이를 물리적 실행 태스크로 이루어진 방향성 비순환 그래프(DAG)로 변환한 다음, 태스크를 실행기에 분배해 실행을 예약합니다.

그런 다음 클러스터 관리자가 배포 환경의 리소스 관리 시스템에 따라 리소스를 할당합니다.

워커 노드는 실행기에 컴퓨팅 리소스를 제공하며, 실행기는 태스크를 수행하고 태스크가 실행되는 동안 데이터를 보관합니다.

Databricks와 같은 일부 관리형 환경에서는 워커 하나당 실행기 하나를 사용하지만, 이는 Spark 전반에 적용되는 규칙은 아닙니다. 클러스터 구성에 따라 워커 하나에서 여러 실행기를 호스팅할 수 있습니다.

2-1. 전자상거래 사용 사례 – 입력 데이터 구조

한 부티크 전자상거래 회사를 가정해 보겠습니다.

이 회사에는 판매 기록 10,000건이 있으며, 각 기록에는 재고 관리 단위(SKU)와 판매 수량이 포함되어 있습니다. 가능한 SKU는 다음 세 가지라고 가정합니다: A, B, 그리고 C.

표 1. 전자상거래 판매 기록 내역 및 집계 예상치 예시.

목표는 SKU별 유효 판매 수량 합계를 계산하는 것입니다:

1. 해당하는 레코드를 제외합니다 qty <= 0, 그런 다음
2. 샘플을 SKU별로 집계합니다.

예시로 전체 데이터 세트에서 필터링 후 10,000건 중 9,800건이 남는다고 가정해 보겠습니다. Spark에서 예상되는 최종 합계는 다음과 같습니다:

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. Spark 클러스터

이 사용 사례에 맞춰 단순화한 클러스터를 설정해 보겠습니다.

워커 노드(워커)

워커 노드 3개를 설정하겠습니다. 각 노드의 인스턴스 유형은 8 vCPU, 메모리 32 GiB입니다:

그림 B. 워커 노드 전체의 리소스 할당 (Kuriko IWAI 제작)

시스템에서 사용할 수 있는 vCPU는 총 24개입니다.

Spark 실행기 및 CPU/메모리 할당

다음과 같이 확보한다고 가정해 보겠습니다.

• 실행기 7개.
• 실행기마다 vCPU 코어 2개.
• 실행기마다 JVM 힙 메모리 3GiB.

Spark 애플리케이션은 클러스터 관리자에게 이 실행기 7개에 리소스를 할당해 달라고 요청합니다.

그러면 클러스터 관리자는 각 워커의 리소스 사용량을 바탕으로 CPU와 메모리 할당량을 결정합니다.

예를 들어, 실행기마다 vCPU 코어 2개를 할당해 실행기 7개에 총 태스크 슬롯 14개를 만들기로 결정할 수 있습니다.

그림 C-1. 워커 A의 리소스 불균형을 반영한 워커별 vCPU 코어, 메모리 및 실행기 할당

실행기에 할당된 메모리 3GiB가 실행기의 전체 메모리는 아니라는 점에 유의하세요. 워커와 기타 실행 작업을 위한 메모리 오버헤드도 확보해야 하기 때문입니다.

단일 태스크에 vCPU 1개가 필요하다고 가정하면 (spark.task.cpus = 1), 각 워커는 다음과 같이 vCPU를 할당합니다.

그림 C-2. 워커별 병렬 vCPU 할당

여기서 두 가지 핵심 사항을 알 수 있습니다.

1. 각 워커의 CPU 사용률은 서로 다릅니다(그림 C-2에서 워커 A의 사용률은 75%인 반면, 워커 B와 C는 50%를 유지합니다).
2. 시스템에는 vCPU 코어가 24개(워커당 vCPU 8개 × 워커 3개) 있지만, 이 태스크에 사용할 수 있는 코어는 14개뿐이며 나머지 10개는 사용되지 않습니다.

2-3. Spark의 기반인 Java 가상 머신(JVM)

Apache Spark는 Java 가상 머신(JVM)을 기반으로 구축되었습니다.

JVM은 Spark의 바이트코드를 실행하는 데 필요한 런타임 환경입니다.

Spark는 Scala/Java로 작성되었으므로 다음과 같습니다.

• 애플리케이션에서 Scala 또는 Java를 사용하는 경우: 코드는 JVM 내부에서 직접 실행됩니다.
• PySpark(Python)를 사용하는 경우: Python 프로세스가 코드를 실행하지만, JVM 기반 드라이버와 Py4J 같은 브리지를 통해 통신합니다. 워커 노드의 Spark 실행기는 여전히 JVM 내부에서 실행되며, JVM은 Python 작업자를 관리해 Python 코드와 Python 사용자 정의 함수(Python UDF)를 실행합니다.

두 경우 모두 JVM은 드라이버와 실행기의 실행 스레드, 메모리 할당 및 가비지 컬렉션을 관리합니다.

PySpark 애플리케이션의 드라이버 및 실행기 JVM

PySpark 애플리케이션에서 Python 드라이버는 JVM에서 실행되는 Spark 드라이버와 통신합니다.

• Python 드라이버: 애플리케이션 수준의 PySpark 코드를 작성하고 실행하는 곳입니다.
• 드라이버 JVM: Spark 실행과 스케줄링을 조정합니다.
• 실행기 JVM: Spark 태스크를 실행하고 실행 리소스를 관리합니다. 태스크에 Python 실행이 필요한 경우 Python 작업자 프로세스가 관여합니다.
• 클라이언트 JVM(클라이언트 프로세스 또는 제출 머신): 로컬 머신이나 엣지/게이트웨이 노드에서 실행되는 JVM입니다.

그림에서 보듯이 클라이언트 JVM이 먼저 드라이버 JVM에 연결되고, 드라이버 JVM은 이어서 실행기 JVM들과 조율합니다.

그림 D. 클라이언트, 드라이버 및 실행기 JVM 간의 전체 관계

이 시점부터 드라이버 JVM은 애플리케이션을 관리하는 Spark 드라이버를 호스팅하고, 실행기 JVM들은 해당 태스크의 실행을 담당하는 개별 실행기들을 호스팅합니다.

각 JVM 드라이버에는 데이터 처리, 캐싱 및 사용자 객체를 효율적으로 처리하기 위한 JVM 힙이 있습니다.

2-4. 복원력 있는 분산 데이터셋(RDD) 자세히 살펴보기

복원력 있는 분산 데이터셋(RDD)은 Spark가 병렬로 처리할 수 있는 불변의 분할된 요소 모음입니다.

RDD는 변환 간의 종속성을 추적하므로, 여러 장애 상황에서 손실된 분할을 Spark가 다시 계산할 수 있습니다.

이 이름에는 세 가지 속성이 담겨 있습니다.

• 복원력: Spark는 어떤 RDD가 어떤 이전 RDD에 의존하는지 나타내는 계보를 기록합니다. 모든 중간 결과의 복제본을 유지하는 대신, 누락된 분할을 재구성할 수 있는 경우가 많습니다.
• 분산: 데이터셋은 Spark가 서로 다른 실행기에서 처리할 수 있는 분할로 표현됩니다. 여러 물리적 머신이 반드시 필요한 것은 아닙니다. 로컬 모드에서도 RDD를 사용할 수 있습니다.
• 데이터셋: RDD에는 관계형 행뿐 아니라 키-값 쌍, 정수, 문자열, 튜플 및 기타 지원되는 요소가 포함될 수 있습니다. RDD API에는 다음과 같은 연산이 포함되며 map, filter, reduceByKey, join, 그리고 aggregateByKey.

RDD는 논리적 컬렉션이자 계산을 추상화한 것이지, 특정 파티션이 특정 작업자에 영구적으로 할당된다는 선언이 아닙니다.

파티션

파티션은 데이터의 논리적 청크이며, 일반적으로 스테이지 내 하나의 태스크에 해당합니다.

파티션 100개를 만든다고 가정해 보겠습니다(초기 파티션을 100개로 설정한 것은 설명을 위한 명시적 선택입니다).

입력 데이터가 대략 고르게 분산되어 있다면, 원본 요소가 10,000개일 때 파티션마다 약 100개씩 들어갑니다:

그림 E. 초기 데이터 분할 100개에 대한 병렬 파티션 할당

3. RDD API와 DataFrame API 비교: 내부 Catalyst 최적화

기본 RDD 예제를 살펴보면 계보, 파티션, 셔플을 이해하는 데 도움이 됩니다.

하지만 일반적인 구조화 ETL에서는 열, 스키마 정보, SQL과 유사한 표현식, 최적화 도구가 파악할 수 있는 연산을 제공하는 DataFrame API를 시작점으로 삼는 것이 좋습니다.

아래 다이어그램은 이들 사이의 관계를 보여 줍니다:

그림 F. Spark 그래프 개념 간의 관계: RDD 계보, DataFrame, Spark 실행 DAG

RDD에서 변환을 정의하기 위해 RDD 계보를 사용하는 것처럼, DataFrame 쿼리 계획은 구조화된 쿼리를 나타냅니다.

Spark는 종속성과 물리 연산자를 바탕으로 실행 작업을 구성합니다.

DataFrame 실행은 내부적으로 RDD 기반 메커니즘을 사용할 수 있지만, 그렇다고 해서 DataFrame 코드를 작성할 때 사용자가 RDD 계보를 직접 구성해야 하는 것은 아닙니다.

3-1. DataFrame 쿼리 계획

DataFrame 쿼리 계획은 실행 엔진이 생성하는 논리적·물리적 단계의 순서입니다(그림 L의 보라색 상자).

DataFrame 쿼리는 여러 표현 단계를 거칩니다:

1. 논리적 계획: 요청된 연산(스캔, 필터링, 그룹화, 집계)에 대한 설명입니다.
2. 분석 및 최적화된 논리적 계획: Spark가 참조를 확인하고 열 가지치기나 조건자 푸시다운과 같은 적용 가능한 논리적 최적화를 수행합니다.
3. 물리적 계획: Spark가 실행 가능한 연산자(예: 스캔, 부분/최종 집계, 교환, 조인, 정렬)를 선택합니다.
4. 실행: 물리적 계획이 Spark의 실행 메커니즘을 통해 수행되며, 최종적으로 실행기의 태스크가 관여합니다.

이 사용 사례에서 물리적 계획에는 다음 단계가 포함될 수 있습니다:

• 부분 해시 집계,
• SKU별 교환, 그리고
• 최종 해시 집계.

3-2. 작업 비교: RDD API와 DataFrame API

SKU 집계를 예로 들어 RDD API와 DataFrame API를 비교해 보겠습니다.

RDD API

요소는 (sku, qty) 쌍입니다. Python 집계 로직을 명시적으로 작성하며, Spark는 RDD 종속성을 따릅니다:

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

DataFrame API

다음과 같다고 가정해 보겠습니다 sales_df 열이 있습니다 sku (문자열) 및 qty (숫자형):

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

여기서 두 예제는 같은 비즈니스 문제를 해결합니다:

유효하지 않은 판매 기록을 필터링하고 유효한 수량을 SKU별로 집계합니다.

그리고 두 경우 모두 collect() Spark 실행을 트리거하고 모든 결과 레코드를 드라이버로 반환합니다.

여기서 핵심적인 차이는 결과의 표현 방식입니다:

• RDD API: Python 요소를 반환합니다. 예를 들면 (sku, total_qty) 튜플입니다.
• DataFrame API: 구조화된 Row 명명된 필드가 있는 객체를 반환합니다. 예를 들면 sku 및 total_qty.

다음은 RDD API와 DataFrame API를 전체적으로 비교한 내용입니다:

표 2. RDD API와 DataFrame API 비교 – 기능 및 추상화 비교: (Kuriko IWAI 작성)

4. 마무리

Spark 튜닝의 목표는 파티션 수나 CPU 사용량을 각각 극대화하는 것이 아닙니다.

불안정성을 초래하지 않으면서 전체 실행 비용과 지연 시간을 줄이는 것이 목표입니다.

이 글에서는 Spark 시스템의 핵심 관계를 살펴봅니다:

• 리소스: 워커는 실행기를 호스팅하고, 실행기는 태스크를 실행하며, 설정된 CPU와 메모리 용량은 동시 실행 수를 제한합니다.
• RDD 프로그래밍: 변환은 불변 RDD 정의와 계보를 생성하고, 액션은 계산을 시작합니다.
• DataFrame 프로그래밍: 변환은 논리적 쿼리 연산을 생성하고, 최적화 과정에서는 실행 가능한 물리 연산자를 선택하며, 액션은 계산을 시작합니다.
• 실행: Spark는 작업을 작업, 스테이지, 태스크로 구성하며, 태스크는 실행기에서 파티션을 처리합니다. 셔플과 기타 종속성은 스테이지 사이에 조정 경계를 만듭니다.

Spark가 분산 워크로드를 실행하는 방식을 이해하는 것은 기초에 불과합니다.

RDD API를 사용하는 7단계 계산 흐름을 알아볼 수 있습니다.