[Spark] Spark Join 종류

스파크에서 Join 시에는 상황에 따라 Shuffle 연산을 수행해야 하고, 이는 executor 사이의 방대한 데이터 이동을 야기한다.

스파크에서 사용되는 조인의 방식은 아래와 같다.

1. Broadcast Hash Join

두 개의 데이터를 조인할 때 한 쪽이 매우 작고 한 쪽은 매우 큰 사이즈의 데이터일 때, 더 작은 쪽의 데이터가 Driver에 의해 모든 executor로 복사되는 방식이다. 

1. 사이즈가 작은 테이블의 키 값을 해시 테이블로 만든다.
2. 해시 테이블의 값을 큰 테이블 파티션에 복사해준다.
3. 각 파티션 내부에서 조인을 진행한다.

이 방식은 shuffle을 생략할 수 있기 때문에 데이터 이동에 따른 코스트가 사라진다.

모든 executor에 데이터를 복사하기 때문에 복사하려는 dataset이 일정 크기보다 더 작아야 broadcast hash join을 사용할 수 있는데, 이 값은 `spark.sql.autoBroadcastJoinThreshold` 에서 설정 가능하다. (default 10MB)

 

 

2. Shuffle Sort Merge Join

이름 그대로 데이터를 shuffle 하고 sort 해서 merge 하는 join 방식이다.

`spark.sql.autoBroadcastJoinThreshold` 깂이 -1로 설정되어 있다면 해당 조인을 먼저 수행한다.

1. Dataset에서 동일한 key를 가진 레코드가 동일 partition에 위치할 수 있게끔 shuffle을 수행한다. (shuffle)
   이부분은 두 데이터셋에서 공통된 "정렬된"컬럼이 저장된 partitioned bucket을 만들면 생략이 가능하다 -> 확인필요
2. executor 내부에서 두 데이터를 key값에 따라 정렬한다 (sort)
3. 정렬된 key값을 돌면서 key가 일치하는 row끼리 join한다. (merge)

 

3. Shuffle Hash Join

이 조인 방법은 Broadcast하지 않는 shuffle hash join이다. 즉, 두 테이블 모두 충분히 작지 않은 경우 수행된다.

마찬가지로  shuffle을 통해 같은 key를 가진 row를 동일한 executor에 배치시킨다. 이후 각 executor에서 hash join을 수행한다.

 

 

sparkSQL 옵티마이저가 알아서 잘 적합한 조인 방법을 찾아주겠지만 사용자가 생각하기에 Broadcast join이 더 적합하다고 생각될 때에는 아래와 같이 Dataframe API를 통해 옵티마이저에게 힌트를 제공할 수 있다.

A.join(broadcast(B), some_condition)

 

 

스파크 조인 전략 관련하여 아래에서 잘 설명되어 있다.

https://blog.clairvoyantsoft.com/apache-spark-join-strategies-e4ebc7624b06