데이터 엔지니어 인터뷰 — Spark

작성일 | 분류 |
Spark의 실행 구조와 지연 평가, 셔플과 조인 전략, 파티션과 병렬도, AQE, 메모리 관리와 OOM, 스트리밍 처리를 질문과 핵심 답변으로 정리합니다.

실행 구조

Driver와 Executor의 역할을 설명해보세요

Driver는 애플리케이션의 제어 주체입니다. 사용자 코드를 실행하며 논리 계획을 만들고, DAG를 Stage로 쪼개 Task를 스케줄링하고, 결과를 모읍니다. Executor는 실제로 Task를 실행하고 데이터를 캐시하는 JVM 프로세스입니다.

collect()가 위험한 이유가 여기 있습니다. 전체 결과를 Driver 메모리로 끌어오기 때문에 데이터가 크면 Driver가 OOM으로 죽습니다. 확인이 목적이면 show()나 take(n)을 씁니다.

Transformation과 Action의 차이는 무엇인가요

Transformation은 새 DataFrame을 정의만 하고 실행하지 않습니다(지연 평가). Action이 호출되는 순간 그때까지 쌓인 계획이 최적화되어 실행됩니다.

df2 = df.filter(...)     # transformation — 아무 일도 안 일어남
df3 = df2.withColumn(...)  # transformation
df3.count()              # action — 여기서 실행

지연 평가 덕분에 Catalyst가 전체 계획을 보고 predicate pushdown, column pruning 같은 최적화를 할 수 있습니다. 대신 에러가 Action 시점에 터지므로 어느 줄이 문제인지 바로 안 보이는 단점이 있습니다.

narrow와 wide transformation을 구분해보세요

narrow는 출력 파티션이 입력 파티션 하나에만 의존합니다(map, filter, union). 셔플이 없고 같은 Stage 안에서 파이프라인으로 처리됩니다.

wide는 여러 입력 파티션의 데이터가 필요합니다(groupBy, join, repartition, distinct). 셔플이 발생하고 Stage 경계가 생깁니다.

Stage 개수를 세려면 셔플 개수를 세면 됩니다.

셔플과 조인

셔플이 비싼 이유는 무엇인가요

디스크 쓰기와 네트워크 전송이 함께 일어나기 때문입니다. map 쪽 Task가 파티션별로 데이터를 정렬·분할해 로컬 디스크에 쓰고, reduce 쪽 Task가 그것을 네트워크로 당겨옵니다. 직렬화·역직렬화 비용도 붙습니다.

게다가 map Stage가 전부 끝나야 reduce가 시작되므로 배리어가 생깁니다.

조인 전략에는 어떤 것들이 있나요

전략 조건 특징
Broadcast Hash Join 한쪽이 충분히 작음 작은 쪽을 전체 Executor에 복제, 셔플 없음
Shuffle Hash Join 한쪽이 해시 테이블로 들어갈 만큼 작음 셔플 후 해시 테이블 구성
Sort Merge Join 양쪽 다 큼 셔플 후 정렬, 대용량 기본값
Broadcast Nested Loop 조인 조건이 등가가 아님 매우 비쌈, 피해야 함

spark.sql.autoBroadcastJoinThreshold(기본 10MB) 이하면 broadcast가 선택됩니다. 통계가 부정확해 broadcast가 안 걸리는 경우 broadcast() 힌트로 강제할 수 있는데, 실제로 크면 Executor OOM이 나므로 크기를 확인하고 써야 합니다.

from pyspark.sql.functions import broadcast
big.join(broadcast(small), "key")

데이터 skew는 어떻게 진단하고 해결하나요

진단은 파티션별 건수 분포를 보면 됩니다.

from pyspark.sql.functions import spark_partition_id, count

df.groupBy(spark_partition_id().alias("pid")) \
  .agg(count("*").alias("rows")) \
  .orderBy("rows", ascending=False).show(10)

해결책은 상황에 따라 다릅니다.

  • salting — 키에 임의 접미사를 붙여 흩뜨린 뒤 2단계로 집계
  • broadcast join — 작은 쪽을 복제해 셔플 자체를 없앰
  • AQE skew join — Spark 3의 spark.sql.adaptive.skewJoin.enabled가 큰 파티션을 자동 분할
  • NULL 키 분리 — NULL이 한 파티션에 몰리는 경우가 흔하므로 미리 걸러내거나 임의 키를 부여

파티션과 병렬도

동시에 실행되는 Task 수는 무엇이 결정하나요

min(전체 Core 수, 파티션 수)입니다. Core는 상한선일 뿐이고 실제 Task 수는 파티션이 정합니다. 파티션이 5개면 Core가 100개여도 5개만 돕니다.

repartition과 coalesce의 차이는 무엇인가요

repartition(n)은 항상 셔플이 일어나며 균등 분배를 보장하고, 파티션을 늘릴 수도 줄일 수도 있습니다. coalesce(n)은 셔플 없이 인접 파티션을 병합하므로 줄이기만 가능하고 균등 분배를 보장하지 않습니다.

coalesce는 어디에 두느냐가 성능을 결정합니다. 계산 앞에 두면 이후 전체가 그 병렬도로 묶이고, 저장 직전에 두면 파일 개수만 줄어듭니다.

# 나쁨 — 이후 계산이 전부 4 병렬로 묶임
df.coalesce(4).filter(...).write.parquet(path)

# 좋음
df.filter(...).coalesce(4).write.parquet(path)

spark.sql.shuffle.partitions는 무엇인가요

셔플 이후 DataFrame의 파티션 수입니다. 기본값 200이라, 작은 데이터에서도 200개 Task가 생겨 오버헤드만 커지는 경우가 많습니다. 반대로 대용량에서는 200이 너무 적어 파티션 하나가 비대해집니다.

Spark 3의 AQE가 런타임 통계를 보고 작은 파티션을 자동 병합해주므로, 예전만큼 손으로 맞출 필요는 줄었습니다.

AQE(Adaptive Query Execution)는 무엇을 하나요

실행 중 수집한 실제 통계로 계획을 다시 짜는 기능입니다. 세 가지가 핵심입니다.

  • 파티션 병합 — 셔플 후 작은 파티션들을 합쳐 Task 수를 줄임
  • 조인 전략 전환 — 런타임 크기를 보고 Sort Merge Join을 Broadcast로 바꿈
  • skew join 처리 — 비대한 파티션을 쪼개 분산

Spark 3.2부터 기본 활성입니다. 명시한 coalesce 값과 실제 파티션 수가 다르게 보이는 원인이기도 합니다.

메모리와 캐시

Executor 메모리 구조를 설명해보세요

크게 Execution 메모리(셔플, 정렬, 조인 해시 테이블)와 Storage 메모리(캐시된 데이터)로 나뉘고, 둘은 통합 영역에서 서로 빌려 씁니다. 여기에 JVM 힙 밖의 오버헤드 영역(spark.executor.memoryOverhead)이 따로 있는데, PySpark 워커와 네트워크 버퍼가 여기를 씁니다.

PySpark에서 OOM이 나는데 힙은 여유 있는 경우, 대개 오버헤드가 부족한 것입니다.

cache와 persist의 차이는 무엇인가요

cache()는 persist(MEMORY_AND_DISK)의 축약입니다. persist()는 저장 레벨을 고를 수 있습니다.

캐시는 여러 번 재사용될 때만 이득입니다. 한 번만 쓰는 DataFrame을 캐시하면 메모리만 낭비합니다. 그리고 캐시도 지연 평가라 Action이 한 번 돌아야 실제로 적재됩니다.

다 쓴 캐시는 unpersist()로 풀어야 다른 작업이 메모리를 쓸 수 있습니다.

OOM이 났을 때 어떻게 접근하나요

어디서 났는지부터 나눕니다.

  • Driver OOM — collect(), toPandas(), 너무 큰 broadcast. 결과를 끌어오지 말고 저장하도록 바꿉니다
  • Executor OOM — 파티션 하나가 너무 큼(skew), 파티션 수 부족, 캐시 과다. 파티션을 늘리거나 skew를 처리합니다
  • 오버헤드 초과 — PySpark UDF나 대량 직렬화. memoryOverhead를 올립니다

executor.cores를 키우면 Task가 힙을 나눠 쓰므로 Task당 메모리가 줄어드는 점도 함께 봐야 합니다.

파일과 포맷

Parquet을 쓰는 이유는 무엇인가요

컬럼너 포맷이라 필요한 컬럼만 읽을 수 있고(column pruning), 컬럼 단위로 압축 효율이 높으며, 통계(min/max)를 담고 있어 조건에 맞지 않는 블록을 건너뛸 수 있습니다(predicate pushdown).

행 단위로 전부 읽어야 하는 CSV/JSON과 비교하면 스캔량 차이가 큽니다.

small files 문제는 왜 생기고 어떻게 다루나요

파티션 수만큼 파일이 생기므로, 파티션이 많고 데이터가 적으면 작은 파일이 대량으로 만들어집니다. 파일마다 메타데이터 조회와 오픈 비용이 붙어 읽기 성능이 떨어지고, 오브젝트 스토리지에서는 요청 수 자체가 비용입니다.

저장 직전에 coalesce로 줄이거나, maxRecordsPerFile로 파일 크기를 제어하거나, 주기적으로 compaction 잡을 돌립니다.

df.write.option("maxRecordsPerFile", 1_000_000).parquet(path)

파티셔닝 컬럼은 어떤 기준으로 고르나요

조회 조건으로 자주 쓰이면서 카디널리티가 적당한 컬럼이 좋습니다. 보통 날짜입니다.

카디널리티가 너무 높으면(예: user_id) 디렉터리가 폭발하고 작은 파일이 양산됩니다. 반대로 너무 낮으면 파티션 프루닝 효과가 없습니다.

스트리밍

Structured Streaming의 전달 보장은 어떻게 되나요

체크포인트에 오프셋과 커밋 로그를 남기고, 실패 시 커밋되지 않은 배치를 같은 batch_id로 재실행합니다. 기본은 at-least-once이고, 싱크가 멱등하면 결과적으로 exactly-once가 됩니다.

체크포인트를 로컬 경로에 두면 클러스터 재시작 시 사라져 전량 재처리가 일어납니다. 반드시 S3 같은 영속 스토리지에 둬야 합니다.

watermark는 무엇인가요

늦게 도착하는 이벤트를 언제까지 기다릴지 정하는 기준입니다. 이벤트 시간 기준 집계에서 상태를 무한정 들고 있을 수 없으므로, watermark보다 오래된 상태는 정리하고 그보다 늦게 온 이벤트는 버립니다.

df.withWatermark("event_time", "10 minutes") \
  .groupBy(window("event_time", "5 minutes")).count()

지연 허용을 늘리면 정확도가 오르지만 상태 크기와 결과 지연도 함께 늘어납니다.

관련 글