Spark & PySpark은(는) 무엇을 하나요?
느린 Spark 작업의 거의 대부분은 세 가지 이유 중 하나로 느립니다: 피할 수 있는 셔플(avoidable shuffle), 쏠린 키(skewed key), 또는 한 번에 한 행씩 Python을 통해 데이터가 누출되는 경우입니다. 비용이 많이 드는 실수는 계획을 읽기 전에 클러스터 크기를 조정하는 것입니다 – 더 많은 익스큐터(executor) 비용을 지불하여 동일한 비효율적인 DAG를 더 빠르게 실행하는 것입니다.
운영 절차
1단계: 입력 수집
- 작업 코드와, 느린 작업의 경우 느린 스테이지의 Spark UI 보기(작업 기간 분포, 셔플 읽기/쓰기 크기) 및
df.explain(True)출력. - 입력 데이터 크기와 형식, 그리고 모든 조인의 양쪽 크기.
- 클러스터 형태: 익스큐터 수, 코어 수, 메모리.
- 작업이 새로운 작성인지 튜닝 패스인지 – 튜닝의 경우 코드를 건드리기 전에 계획에서 진단합니다.
2단계: 올바른 세션에서 시작
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = (
SparkSession.builder
.appName("etl")
.config("spark.sql.shuffle.partitions", "200")
.config("spark.sql.adaptive.enabled", "true")
.getOrCreate()
)
적응형 쿼리 실행(AQE)을 활성화합니다. AQE는 런타임에 셔플 파티션을 병합하고 조인 전략을 전환합니다. 그런 다음 spark.sql.shuffle.partitions를 의도적으로 설정합니다: 셔플 파티션당 100-200MB를 목표로 하므로, partitions ≈ 셔플 데이터 크기 ÷ 128MB이고, 총 익스큐터 코어의 최소 2-3배가 되어 코어가 유휴 상태가 되지 않도록 합니다.
3단계: DataFrame API 안에 머물고 Python을 벗어나기
DataFrame 연산은 Catalyst와 Tungsten을 통해 쿼리 최적화와 off-heap 메모리를 제공합니다. RDD는 DataFrame 기본 요소가 없는 경우에만 사용합니다. Python UDF는 각 행을 Python 워커로 직렬화하고 다시 되돌려 코드 생성을 차단합니다. 선호 순서:
- 내장
pyspark.sql.functions(예:F.regexp_extract,F.when). - Pandas (벡터화) UDF – Python이 불가피할 때 사용.
- 스칼라 Python UDF – 최후의 수단으로만 사용.
from pyspark.sql.functions import pandas_udf
@pandas_udf("double")
def normalize(s):
return (s - s.mean()) / s.std()
4단계: 크기에 따라 조인 전략 선택
- 작은 쪽이 메모리에 들어갈 때는 브로드캐스트합니다. Spark는
spark.sql.autoBroadcastJoinThreshold(기본값 10MB) 이하를 자동으로 브로드캐스트합니다. 익스큐터 메모리가 허용할 때 수백 MB까지의 테이블은 명시적으로 브로드캐스트하며, ~1GB를 실질적인 상한선으로 간주합니다 – 브로드캐스트는 테이블을 모든 익스큐터에 복사하므로 잘못 판단하면 작업이 OOM됩니다.
from pyspark.sql.functions import broadcast
result = large.join(broadcast(small), "key")
- 조인 전에 열을 필터링하고 프로젝션하여 셔플을 줄입니다. 모든 드롭된 열과 행은 절약된 셔플 바이트입니다.
- 두 개의 큰 테이블 → 소트-머지 조인을 수용하지만, 먼저 쏠림을 확인합니다(5단계).
5단계: 쏠림 진단 및 수정
쏠림 증상: 하나나 두 개의 작업을 제외한 스테이지가 완료됨; 최대 작업 기간이 중앙값의 ~3-5배를 초과함; Spark UI에서 하나의 셔플 읽기 파티션이 나머지보다 훨씬 큼. AQE의 쏠림 조인 처리(skew-join handling)는 파티션이 중앙값 파티션 크기의 5배를 초과하고 256MB(기본값)를 초과할 때 분할합니다 – 계획에서 실행되었는지 확인합니다. AQE가 도움이 되지 않을 때(예: 쏠린 집계), 핫 키(hot key)에 소금(salt)을 뿌립니다: 큰 쪽의 키에 0..N 범위의 임의 접미사를 추가하고, 작은 쪽을 모든 N개의 접미사로 확장한 후 조인한 다음 소금을 제거합니다. N을 쏠림 계수에 가깝게 선택합니다(중앙값의 20배인 키는 N=20).
6단계: 파티셔닝 및 출력 레이아웃 제어
repartition(n, col)은 데이터를 균형 있게 맞추기 위해 전체 셔플을 수행합니다. 넓은 쓰기 전에 사용합니다.coalesce(n)은 셔플 없이 파티션 수를 줄입니다. 아주 작은 출력 파일을 피하기 위해 사용합니다.- 출력 파일의 목표 크기는 128MB-1GB입니다. 수천 개의 작은 파일은 모든 다운스트림 리더에 부담을 줍니다.
- 출력은 카디널리티가 낮은 열(날짜, 지역 – 수십에서 수백 개의 값, ID는 절대 안 됨)로만 파티션합니다:
df.write.partitionBy("dt").mode("overwrite").parquet("/data/out")
7단계: 재사용되는 것만 캐시
DataFrame이 DAG에서 여러 번 사용될 때만 캐시하고, 구체화한 후 사용이 끝나면 unpersist합니다:
df.cache()
df.count() # materialize
# ... reuse df ...
df.unpersist()
한 번만 사용되는 DataFrame을 캐시하면 메모리를 낭비하고 중요한 데이터를 축출(evict)할 수 있습니다.
8단계: 계획으로 확인
df.explain(True)를 다시 실행하여 의도한 전략이 나타나는지 확인합니다(BroadcastHashJoin, 푸시된 필터, 제거된 Exchange가 없는지). 컬럼 형식(Parquet, ORC)을 읽어 조건자 푸시다운(predicate pushdown)이 작동하도록 하고, 큰 데이터에 대해 collect()를 절대 사용하지 말며, 체이닝 시 문자열 열보다 F.col 참조를 선호합니다.
작업 예제: 나쁜 조인 vs 좋은 조인
나쁜 예 – 2TB 이벤트 테이블이 200MB 차원 테이블에 조인되고, 전체 너비, Python UDF로 파싱:
result = (events.join(dims, "store_id") # sort-merge: 2TB 전체 셔플
.withColumn("region", parse_region_udf("meta")) # 행 단위 Python
.filter(F.col("dt") == "2024-06-01")) # 조인 후 필터
계획은 양쪽에 Exchange를 보여주고(2TB 셔플됨), UDF가 코드 생성을 차단합니다.
좋은 예 – 먼저 필터링하고 프로젝션, 차원을 브로드캐스트, 내장 함수 사용:
result = (events
.filter(F.col("dt") == "2024-06-01") # 먼저 ~30GB로 축소
.select("store_id", "amount", "meta")
.join(broadcast(dims.select("store_id", "region_code")), "store_id")
.withColumn("region", F.regexp_extract("meta", r"region=(\w+)", 1)))
계획은 이제 큰 쪽의 Exchange 없이 BroadcastHashJoin을 보여줍니다: ~30GB 스캔, 큰 테이블의 셔플 제로, Python 경계 없음.
산출물
수정된 작업 코드와 함께 튜닝 노트를 작성합니다: 각 조인에 대한 조인 전략과 크기 증거, 셔플-파티션 설정과 그 계산, 발견된 모든 쏠림(최대 대 중앙값 작업 비율)과 적용된 수정, 캐싱 결정, 출력 파일 크기 예상, 그리고 튜닝 패스의 경우 이전/이후 explain 차이.
금지 사항
- 크기를 확인하지 않은 테이블을 브로드캐스트하지 마십시오 – 잘못 판단된 브로드캐스트는 모든 익스큐터를 동시에 OOM시킵니다.
pyspark.sql.functions을 확인하기 전에 Python UDF를 작성하지 마십시오; 거의 모든 문자열, 날짜 및 조건부 로직에 내장 함수가 있습니다.- 큰 DataFrame에 대해
collect()또는toPandas()를 사용하지 마십시오 – 클러스터의 데이터를 하나의 드라이버로 모읍니다. - 출력을 카디널리티가 높은 열로 파티션하지 마십시오 – 수백만 개의 디렉토리, 각각 한 행씩.
- 느린 스테이지가 하나의 쏠린 작업인 작업에 익스큐터를 추가하지 마십시오 – 더 많은 코어로 하나의 파티션을 나눌 수 없습니다.
- 반사적으로 캐시하지 마십시오 – 다운스트림에서 재사용되는 DataFrame만 캐시하고, unpersist하십시오.
품질 기준
완료된 작업은 다음 조건을 모두 충족해야 합니다: 모든 조인이 그 전략과 크기 증거를 명시함; spark.sql.shuffle.partitions가 데이터 크기에서 파생되고 기본값으로 남지 않음; Spark UI가 셔플 스테이지에서 중앙값 기간의 ~3배 이상인 작업을 보여주지 않음(또는 쏠림이 설명되고 수정됨); 내장 함수가 존재하는 곳에 스칼라 Python UDF가 남아 있지 않음; 출력 파일이 128MB-1GB 범위에 속함; 최종 explain이 의도한 계획을 확인함.