Spark 배치 처리

학습 목표

Spark가 분산 배치 처리에서 process 레이어를 담당하는 방식을 설명할 수 있다.

transformation과 action·lazy 실행을 구분할 수 있다.

partition·shuffle이 병렬성·비용에 미치는 영향을 설명할 수 있다.

RDD와 DataFrame API 선택 기준을 말할 수 있다.

일일 집계·backfill 배치 job 초안을 설계할 수 있다.

문제 상황

• 어제 parquet 10TB — 단일 서버 pandas로 OOM, cron 12시간 초과
• groupBy(user_id) 후 디스크 폭발 — shuffle이 뭔지 모름
• Spark job 로그에 200 stages — count()마다 전체 재실행
• partition 1개로 읽고 200 executor — 199개 idle
• Kafka에서 consume은 하는데 어제 slice 집계는 배치로 — Structured Streaming과 역할 혼동
• 4편에서 topic·partition을 봤지만 — Spark 쪽 partition·shuffle이 별개임

이번 편은 bounded 입력을 분산으로 읽고 변환·집계하는 Spark 배치다. Structured Streaming은 6편.

1. Spark가 process 레이어에서 하는 일

3편 배치 모델 — collect → store → process → publish. Spark는 process의 대표 엔진이다.

역할	Spark가 해주는 것
분산 읽기	parquet·JSON·JDBC·Kafka offset range를 partition으로 나눠 읽기
변환·집계	filter·join·groupBy를 클러스터에서 병렬 실행
장애 복구	task 실패 시 lineage로 재계산 (RDD·DAG)
스케일 아웃	executor 추가로 throughput 확장 (shuffle 병목 제외)

S3 raw/2026-07-28/*.parquet
        │
        ▼
Spark batch job (driver + executors)
        │
        ▼
warehouse user_day_agg · ML training set

• Driver — job·stage 계획, 스케줄, action 결과 수집
• Executor — task 실행 — partition 청크 단위
• Cluster manager — YARN·K8s·Standalone — 리소스 할당
• ingest는 Kafka (4편) — Spark는 이미 쌓인 raw 또는 topic 구간을 배치로 읽을 수 있음

2. RDD vs DataFrame

	RDD	DataFrame / Dataset
타입	RDD[T] — 객체 분산 컬렉션	named column·스키마
최적화	개발자가 partition 직접	Catalyst optimizer
API	map·reduceByKey	SQL·groupBy·DataFrame DSL
용도	레거시·비정형·세밀 제어	일반 ETL·SQL 팀 협업

# code/batch_job.example.py (발췌) — DataFrame API 권장
orders = spark.read.parquet("s3://lake/raw/orders/dt=2026-07-28")
daily = (
    orders.filter(col("status") == "PAID")
    .groupBy("user_id")
    .agg(sum("amount").alias("revenue"))
)
daily.write.mode("overwrite").parquet("s3://lake/gold/user_day_revenue/dt=2026-07-28")

• Dataset[T] — typed row — Scala·Java에서 강타입
• Spark SQL — spark.sql("SELECT ...") — 동일 엔진
• RDD — sc.textFile·map — 신규 프로젝트는 DataFrame 우선
• UDF — Catalyst 밖 — 성능·vectorization 손해 — 표현식·built-in 우선

3. Transformation vs Action

Spark는 lazy — transformation은 계획(DAG) 만 쌓고, action이 실행을 트리거한다.

종류	예	실행
Transformation	filter, select, join, groupBy	지연 — lineage 추가
Action	count, collect, write, show	즉시 job 제출

read parquet  →  filter  →  groupBy  →  write parquet
   (T)            (T)         (T)         (A) ← job runs here

• 여러 action — DAG 재사용 안 되면 각각 full scan — cache·한 번에 write
• debug — df.explain() — physical plan·shuffle 유무
• collect() — driver로 전부 끌어옴 — 대용량 금지
• write — action — partition 수 = 출력 파일 수 힌트

# 나쁜 예 — action 두 번
n = daily.count()          # full job 1
daily.write.parquet(path)  # full job 2

# better — write만, row count는 별도 메트릭 또는 write 결과
daily.write.parquet(path)

4. Partition

Partition — Spark가 나눠 읽고 쓰는 병렬 단위. Kafka partition(4편)과 이름만 같음.

개념	설명
Input partition	HDFS block·parquet 파일·maxPartitionBytes로 나뉜 읽기 청크
RDD/DataFrame partition	메모리·디스크 상 논리 분할 — task 1개 ≈ partition 1개 (map stage)
repartition(n)	shuffle — partition 수 변경
coalesce(n)	shuffle 없이 줄이기 — 파일 merge에 유용

• 너무 적음 — CPU 미활용 — executor 많아도 idle
• 너무 많음 — task 오버헤드·작은 파일 난립
• Rule of thumb — executor 코어 × 2~4 정도 partition (입력·shuffle 후 조정)
• Kafka 배치 read — assign offset range — consumer partition과 Spark partition 매핑 설계

# code/partition_shuffle.example.py (발췌)
# skew 의심 시 salt key — shuffle 전
from pyspark.sql.functions import concat, col, lit, rand

salted = orders.withColumn("salt_key", concat(col("user_id"), lit("_"), (rand() * 10).cast("int")))
# ... groupBy salt_key 후 다시 user_id 로 reduce

5. Shuffle

Shuffle — 같은 key를 같은 task로 모으는 전역 데이터 이동. 비용이 큰 연산.

연산	shuffle
filter, map, select	없음 (narrow)
groupBy, join, repartition, distinct	있음 (wide)
orderBy (전역)	있음

Stage 1 (map):  partition 0..N  — local filter
        │
   shuffle write / read  (disk · network)
        │
Stage 2 (reduce): groupBy user_id

• spark.sql.shuffle.partitions — 기본 200 — 데이터 크기에 맞게 (예: 50~2000)
• Shuffle spill — memory 부족 시 디스크 — 느림 — partition 수·메모리 튜닝
• Skew — 특정 key에 데이터 편중 — salting·AQE skew join (Spark 3+)
• Broadcast join — 작은 테이블 전 executor 복사 — shuffle 회피

# code/spark-job.conf.yaml (발췌)
spark:
  executor.instances: 8
  executor.cores: 4
  executor.memory: 8g
  sql.shuffle.partitions: 64
  sql.adaptive.enabled: true

6. 배치 job 설계 체크리스트

#	질문
1	입력이 bounded slice인가 (dt·offset range)
2	action은 최소화·cache 필요한가
3	groupBy·join shuffle 비용을 explain으로 확인했는가
4	출력 partition·파일 크기가 lake 권장(128MB~1GB)에 맞는가
5	job 멱등인가 — 같은 dt overwrite·partition replace
6	실패 시 재실행·backfill 경로가 있는가

# bounded slice — daily batch
dt = "2026-07-28"
raw = spark.read.parquet(f"s3://lake/raw/orders/dt={dt}")
gold = raw.groupBy("user_id").agg(sum("amount").alias("revenue"))
gold.write.mode("overwrite").parquet(f"s3://lake/gold/user_day/dt={dt}")

• Incremental — dt partition append vs merge — 7편 ETL·9편 품질과 연결
• Kafka → Spark batch — startingOffsets·endingOffsets로 하루 구간 bounded 처리
• 비용 — shuffle 많은 job은 야간·spot executor — 2편 scale-out

정리

• Spark 배치 — bounded raw를 분산 process → gold table
• DataFrame — 일반 ETL·Catalyst 최적화; RDD는 예외적
• Transformation lazy — action이 job 실행
• Partition — 병렬 단위; shuffle — wide 연산·병목
• Kafka partition·consumer group(4편)과 Spark partition은 별개 — 설계 시 둘 다 맞출 것

다음에 다룰 것

• Spark Structured Streaming
• watermark, trigger, sink, micro-batch vs continuous

---

해당 내용은 Spark Documentation (Batch Guide), Kleppmann — Designing Data-Intensive Applications (Ch 10) 를 기반으로 합니다.