Spark

Spark가 고를 수 있는 4가지 물리적 조인 전략

논리적으로 같은 join이라도 Spark는 실행 시점에 네 가지 물리적 전략 중 하나를 고른다. (1) Broadcast Hash Join — 한쪽 테이블이 충분히 작으면(spark.sql.autoBroadcastJoinThreshold, 기본 10MB 미만) 그 테이블 전체를 모든 executor에 복사해 셔플 자체를 없앤다. 가장 빠르지만 작은 쪽이 실제로 메모리에 들어갈 만큼 작아야 한다. (2) Sort-Merge Join — 양쪽을 조인 키로 정렬한 뒤 병합하는 기본 전략. 양쪽 다 크고 브로드캐스트가 불가능할 때 기본으로 선택된다(spark.sql.join.preferSortMergeJoin, 기본 true). (3) Shuffled Hash Join — 정렬 없이 한쪽으로 해시 맵을 만들어 조인한다. 정렬 비용을 아낄 수 있지만 해시 맵이 파티션당 메모리에 다 들어가야 한다(AQE의 maxShuffledHashJoinLocalMapThreshold로 유도 가능, 모듈 41 참고). (4) Shuffle-and-Replicate Nested Loop Join — 동등 조인 조건(equi-join, key = key)이 아닌 조인(예: 범위 조인 a.x BETWEEN b.lo AND b.hi)에서 다른 방법이 없을 때 쓰는 최후 수단으로, 양쪽을 전부 비교하므로 가장 느리다.

조인 힌트로 자동 판단을 오버라이드하기

Spark의 자동 판단(테이블 크기 추정 기반)이 항상 맞는 것은 아니다 — 특히 복잡한 변환을 거친 뒤라 통계가 부정확하거나, 외부 소스 통계가 없을 때 오판할 수 있다. 이럴 때 SQL 힌트나 DataFrame API의 .hint()로 원하는 전략을 직접 지정할 수 있다.

SQL: `SELECT /*+ BROADCAST(r) */ * FROM src s JOIN records r ON s.key = r.key` / PySpark: `spark.table("src").join(spark.table("records").hint("broadcast"), "key")`. 힌트 종류는 BROADCAST, MERGE(sort-merge 강제), SHUFFLE_HASH, SHUFFLE_REPLICATE_NL 네 가지이며, 여러 힌트가 충돌하면 이 순서로 우선순위가 적용된다: BROADCAST > MERGE > SHUFFLE_HASH > SHUFFLE_REPLICATE_NL. 즉 BROADCAST 힌트를 준 조인은 다른 힌트나 자동 판단보다 항상 우선해서 적용된다.

힌트를 준 결과가 실제로 의도한 전략으로 실행됐는지는 spark-learn-32에서 다룬 방식대로 .explain()의 물리 실행 계획에서 BroadcastHashJoin/SortMergeJoin/ShuffledHashJoin 문자열을 직접 확인하는 것이 유일하게 확실한 검증 방법이다 — 힌트를 줬다고 항상 그대로 적용되는 것은 아니다(예: BROADCAST 힌트를 줬는데 그 테이블이 실제로는 너무 커서 메모리에 안 들어가면 OOM 위험이 있어 Spark가 무시하거나 다른 방식을 택할 수 있다).