Spark

대용량 테이블 조인 — Sort-Merge Join과 버킷팅

두 개의 매우 큰 테이블을 조인할 때 Spark는 보통 Sort-Merge Join을 사용한다. 먼저 두 테이블을 조인 키 기준으로 셔플해 같은 키를 가진 행들을 같은 파티션에 모으고, 각 파티션 내부를 정렬한 뒤 병합한다. 두 테이블을 조인 키로 미리 버킷팅(bucketing)하고 정렬해두면, 조인 시점에 셔플과 정렬 단계를 아예 건너뛸 수 있어 훨씬 효율적이다.

Driver로 데이터를 반환하지 않는 액션

saveAsTextFile(), write.parquet(), write.jdbc() 같은 액션은 결과를 Driver로 모으지 않는다. 대신 각 Executor가 자신이 처리한 파티션을 S3/HDFS/DB 같은 외부 저장소에 직접 쓴다. 이는 전체 클러스터의 결합된 대역폭을 활용할 수 있어, 데이터를 Driver로 먼저 모으는 방식보다 훨씬 효율적이다.

JDBC 소스 병렬 읽기

spark.read.format("jdbc")로 연결 URL, 테이블명, 인증 정보를 지정해 읽는다. 단일 커넥션으로 테이블 전체를 읽으면 그 커넥션이 병목이 되므로, partitionColumn, lowerBound, upperBound, numPartitions를 함께 지정해 Spark가 여러 동시 커넥션을 열어 데이터를 병렬로 읽도록 해야 한다.

reduceByKey vs groupByKey

reduceByKey는 셔플이 일어나기 전에 각 Executor 로컬에서 값을 먼저 합치는 'map-side combine'을 수행하기 때문에 네트워크로 전송되는 데이터량이 크게 줄어 효율적이다. groupByKey는 이런 로컬 사전 집계 없이 모든 레코드를 그대로 셔플하므로, 한 키에 레코드가 수백만 건 몰리면 OOM으로 이어지기 쉽다.

다만 이 비교는 RDD API에 해당하는 이야기다. DataFrame/Dataset API의 groupBy().agg(...)는 Catalyst가 내부적으로 부분 집계(partial aggregation)를 자동으로 적용하므로, RDD의 groupByKey처럼 무조건 모든 레코드를 셔플하지는 않는다. 'groupByKey를 피하라'는 조언은 RDD 저수준 API에 국한된 것이지 DataFrame의 groupBy에 그대로 적용되지는 않는다는 점을 구분해야 한다.

Dynamic Allocation

Dynamic Allocation은 워크로드에 따라 애플리케이션에 배정되는 Executor 수를 자동으로 늘리거나 줄이는 기능이다. 대기 중인 태스크가 쌓이면 클러스터 매니저에 Executor를 추가 요청하고, Executor가 일정 시간 유휴 상태면 반납한다. 여러 애플리케이션이 자원을 공유하는 클러스터에서 유휴 자원 낭비를 막는 데 중요하다.

Small File Problem 대응

Small File Problem은 Spark가 스토리지에 수천 개의 작은 파일을 써서 이후 읽기 성능을 크게 떨어뜨리는 문제다. 대응책으로는 쓰기 전에 coalesce(n)/repartition(n)으로 출력 파티션 수를 줄이는 방법, 잡 종료 후 별도 컴팩션 유틸리티로 작은 파일을 합치는 방법이 있다. Spark 3.0+에서는 AQE가 셔플 단계에서 작은 파티션을 자동으로 병합해주는 기능도 이 문제를 완화하는 데 도움이 된다.

Speculative Execution

Speculative Execution은 특정 태스크가 평균보다 눈에 띄게 느리게 실행될 때(하드웨어 결함이나 특정 노드의 네트워크 문제 등으로) 다른 노드에 동일 태스크의 복사본을 추가로 실행하는 헬스체크성 메커니즘이다. 먼저 끝나는 쪽의 결과를 채택하고 나머지는 종료한다. spark.speculation은 기본적으로 false(비활성)이므로, 이 기능을 쓰려면 명시적으로 켜야 한다는 점은 원문에 빠져 있는 실무상 유의점이다.

UDF의 비용과 대안

PySpark에서는 일반 Python 함수를 작성한 뒤 pyspark.sql.functions.udf(func, returnType)로 등록해 사용한다. 하지만 UDF는 JVM 바깥의 Python 프로세스에서 실행되기 때문에 Spark가 데이터를 직렬화해 Python으로 보내고 결과를 다시 받아와야 하는 오버헤드가 크다. 가능하면 Scala로 작성되어 JVM에서 직접 실행되는 내장 Spark SQL 함수를 우선 사용해야 하며, 부득이 Python UDF가 필요하다면 Arrow 기반 벡터화 처리를 하는 pandas UDF(Pandas Function API)가 일반 UDF보다 오버헤드가 훨씬 적다는 점도 함께 알아둘 만하다.

checkpoint() vs cache()

checkpoint()는 RDD를 안정적인 스토리지에 저장하면서 lineage를 완전히 제거한다. 이는 lineage를 그대로 유지한 채 데이터만 캐싱하는 cache()와 다르다. 스트리밍이나 GraphX/ALS 같은 반복 알고리즘처럼 DAG가 매우 커질 수 있는 워크로드에서는, lineage가 지나치게 길어져 스택 오버플로우가 발생하거나 장애 복구 시 처음부터 재계산해야 하는 시간이 과도해질 수 있어 checkpoint()가 필수적이다.

컬럼의 고유값 개수 세기

df.select("column").distinct().count()는 distinct라는 wide 트랜스포메이션을 수행한 뒤 행 수를 센다. countDistinct("column")는 최적화된 집계 함수로 고유값 개수를 계산한다. 정확한 값이 꼭 필요하지 않은 매우 큰 데이터셋이라면 HyperLogLog 기반 근사 알고리즘을 쓰는 approx_count_distinct()로 훨씬 빠르게 근사치를 얻을 수 있다.