[Spark] 조인(join) 표현식, 타입 정리

[Spark The Definitive Guide - BIG DATA PROCESSING MADE SIMPLE] 책을 중심으로 스파크를 개인 공부를 위해 요약 및 정리해보았습니다.
다소 복잡한 설치 과정은 도커에 미리 이미지를 업로해 놓았습니다. 즉, 도커 이미지를 pull하면 바로 스파크를 사용할 수 있습니다.

도커 설치 및 활용하는 방법 : [Spark] 빅데이터와 아파치 스파크란 - 1.2 스파크 실행하기
도커 이미지 링크 : https://hub.docker.com/r/ingu627/hadoop
예제를 위한 데이터 링크 : FVBros/Spark-The-Definitive-Guide
예제에 대한 실행 언어는 스칼라(scala)로 했습니다.
앞으로 스파크와 관련된 내용은 딥러닝 부분을 제외하고 스칼라로 진행될 예정입니다.

기본 실행 방법
1. 예제에 사용 될 데이터들은 도커 이미지 생성 후 spark-3.3.0 안의 하위 폴더 data를 생성 후, 이 폴더에 추가합니다.
1.1 데이터와 도커 설치 및 활용하는 방법은 위에 링크를 남겼습니다.
2. 프로그램 시작은 cd spark-3.3.0 후, ./bin/spark-shell 명령어를 실행하시면 됩니다.

1. 조인 표현식

• 스파크는 왼쪽과 오른쪽 데이터셋에 있는 하나 이상의 키값을 비교하고 왼쪽 데이터셋과 오른쪽 데이터셋의 결합 여부를 결정하는 조인 표현식(join expression)의 평가 결과에 따라 2개의 데이터셋을 조인한다.

2. 조인 타입

• 조건 표현식은 두 로우의 조인 여부를 결정하는 반면, 조인 타입은 결과 데이터셋에 어떤 데이터가 있어야 하는지 결정한다.
• 스파크에서 사용할 수 있는 조인 타입은 다음과 같다.
  • 내부 조인(inner join) : 왼쪽과 오른쪽 데이터셋에 키가 있는 로우를 유지
  • 외부 조인(outer join) : 왼쪽이나 오른쪽 데이터셋에 키가 있는 로우를 유지
  • 왼쪽 외부 조인(left outer join) : 왼쪽 데이터셋에 키가 있는 로우를 유지
  • 오른쪽 외부 조인(right outer join) : 오른쪽 데이터셋에 키가 있는 로우를 유지
  • 왼쪽 세미 조인(left semi join) : 왼쪽 데이터셋의 키가 일치하는 왼쪽 데이터셋만 유지
  • 왼쪽 안티 조인(left semi join) : 왼쪽 데이터셋의 키가 오른쪽 데이터셋에 없는 경우에는 키가 일치하지 않는 왼쪽 데이터셋만 유지
  • 자연 조인(natural join) : 두 데이터셋에서 동일한 이름을 가진 컬럼을 암시적(implicit)으로 결합하는 조인을 수행
  • 교차 조인(cross join) : 왼쪽 데이터셋의 모든 로우와 오른쪽 데이터셋의 모든 로우르 조합

• 우선 예제에서 사용할 몇 가지 간단한 데이터셋을 만든다.

val person = Seq(
    (0, "Bill Chambers", 0, Seq(100)),
    (1, "Matei Zaharia", 1, Seq(500, 250, 100)),
    (2, "Michael Armbrust", 1, Seq(250, 100))
).toDF("id", "name", "graduate_program", "spark_status")

val graduateProgram = Seq(
    (0, "Masters", "School of Information", "UC Berkeley"),
    (2, "Masters", "EECS", "UC Berkeley"),
    (1, "PH.D", "EECS", "UC Berkeley")
).toDF("id", "degree", "department", "school")

val sparkStatus = Seq(
    (500, "Vice President"),
    (250, "PMC Member"),
    (100, "Contributor")
).toDF("id", "status")

person.createOrReplaceTempView("person")
graduateProgram.createOrReplaceTempView("graduateProgram")
sparkStatus.createOrReplaceTempView("sparkStatus")

2.1 내부 조인

• 내부 조인(inner join)은 DataFrame이나 테이블에 존재하는 키를 평가한다.
• 그리고 true로 평가되는 로우만 결합한다.
• 내부 조인은 기본 조인 방식으로 JOIN 표현식에 왼쪽 DataFrame과 오른쪽 DataFrame을 지정하기만 하면 된다.

val joinExpression = person.col("graduate_program") === graduateProgram.col("id")

person.join(graduateProgram, joinExpression).show()

• 실행결과

  

• join 메서드의 세 번째 파라미터(joinType)로 조인 타입을 명확하게 지정할 수도 있다.

var joinType = "inner"

person.join(graduateProgram, joinExpression, joinType).show()

• 실행결과

  

2.2 외부 조인

• 외부 조인(outer join)은 DataFrame이나 테이블에 존재하는 키를 평가하여 true나 false으로 평가한 로우를 포함하여 조인한다.
• 왼쪽이나 오른쪽 DataFrame에 일치하는 로우가 없다면 스파크는 해당 위치에 null을 삽입한다.

var joinType = "outer"

person.join(graduateProgram, joinExpression, joinType).show()

• 실행결과

  

2.3 왼쪽 외부 조인

• 왼쪽 외부 조인(left outer join)은 DataFrame이나 테이블에 존재하는 키를 평가한다.
• 그리고 왼쪽 DataFrame의 모든 로우와 왼쪽 DataFrame과 일치하는 오른쪽 DataFrame의 로우를 함께 포함한다.

var joinType = "left_outer"

graduateProgram.join(person, joinExpression, joinType).show()

• 실행결과

  

2.4 오른쪽 외부 조인

• 오른쪽 외부 조인(right outer join)은 DataFrame이나 테이블에 존재하는 키를 평가한다.
• 그리고 오른쪽 DataFrame의 모든 로우와 오른쪽 DataFrame과 일치하는 왼쪽 DataFrame의 로우를 함께 포함한다.

var joinType = "right_outer"

person.join(graduateProgram, joinExpression, joinType).show()

• 실행결과

  

2.5 왼쪽 세미 조인

• 왼쪽 세미 조인(left semi join)은 기존 조인 기능과는 달리 DataFrame의 필터 정도로 볼 수 있다.
  • 두 번째 DataFrame은 값이 존재하는지 확인하기 위해 값만 비교하는 용도로 사용한다.

var joinType = "left_semi"

graduateProgram.join(person, joinExpression, joinType).show()

• 실행결과

  

2.6 왼쪽 안티 조인

• 안티 조인(anti join)은 SQL의 NOT IN과 같은 스타일의 필터로 볼 수 있다.
  • 두 번째 DataFrame의 어떤 값도 포함하지 않고, 단지 값이 존재하는지 확인하기 위해 값만 비교하는 용도로 사용한다.
  • 하지만, 두 번째 DataFrame에 존재하는 값을 유지하는 대신 두 번째 DataFrame에서 관련된 키를 찾을 수 없는 로우만 결과에 포함한다.

var joinType = "left_anti"

graduateProgram.join(person, joinExpression, joinType).show()

• 실행결과

  

2.7 자연 조인

• 자연 조인(natural join)은 조인하려는 컬럼을 암시적으로(implicitly)으로 추정한다.
• 즉, 일치하는 컬럼을 찾고 그 결과를 반환한다.
• 왼쪽과 오른쪽, 그리고 외부 자연 조인을 사용할 수 있다.

2.8 교차 조인

• 교차 조인(cross join)은 조건절을 기술하지 않은 내부 조인을 의미한다.
• 교차 조인은 왼쪽 DataFrame의 모든 로우를 오른쪽 DataFrame의 모든 로우와 결합한다.

person.crossJoin(graduateProgram).show()

• 실행결과

  

3. 스파크의 조인 수행 방식

• 스파크가 조인을 수행하는 방식을 이해하기 위해서는 실행에 필요한 2가지 핵심 전략을 이해해야 한다.
  • 노드간 네트워크 통신 전략
  • 노드별 연산 전략

3.1 네트워크 통신 전략

• 스파크는 조인 시 2가지 클러스터 통신 방식을 활용한다.
  • 전체 노드 간 통신을 유발하는 셔플 조인(shuffle join)과 그렇지 않은 브로드캐스트 조인(broadcast join)이다.
• 이제부터는 사용자가 스파크에서 사용하는 테이블의 크기가 아주 크거나 아주 작다고 가정한다.

큰 테이블과 큰 테이블 조인

• 하나의 큰 테이블을 다른 큰 테이블과 조인하면 Figure 8.1과 같이 셔플 조인이 발생한다.

• 셔플 조인은 전체 노드 간 통신이 발생한다. 그리고 조인에 사용한 특정 키나 키 집합을 어떤 노드가 가졌는지에 따라 해당 노드와 데이터를 공유한다.
  • 이런 통신 방식 때문에 네트워크는 복잡해지고 많은 자원을 사용한다.
• 즉, 전체 조인 프로세스가 진행되는 동안 데이터 파티셔닝 없이 모든 워커 노드에서 통신이 발생함을 의미한다.

큰 테이블과 작은 테이블 조인

• 테이블이 단일 워커 노드의 메모리 크기에 적합할 정도로 충분히 작은 경우 조인 연산을 최적화할 수 있다.
• 브로드캐스트 조인 방법은 작은 DataFrame을 클러스터의 전체 워커 노드에 복제하는 것을 의미한다.
  • 이렇게 하면 조인 프로세스 내내 전체 노드가 통신하는 현상을 방지할 수 있다.
• 다음 Figure 8.2같이 시작 시 단 한 번만 복제가 수행되며 그 이후로는 개별 워커가 다른 워커 노드를 기다리거나 통신할 필요 없이 작업을 수행할 수 있다.

• 브로드캐스트 조인은 이전 조인 방식과 마찬가지로 대규모 노드 간 통신이 발생한다.
• 하지만 그 이후로는 노드 사이에 추가적인 통신이 발생하지 않는다.
• 따라서 모든 단일 노드에서 개별적으로 조인이 수행되므로 CPU가 가장 큰 병목 구간이 된다.
• 다음 코드에서 볼 수 있듯이 스파크가 자동으로 데이터셋을 브로드캐스트 조인으로 설정한 것을 알 수 있다.

val joinExpr = person.col("graduate_program") === graduateProgram.col("id")

person.join(graduateProgram, joinExpr).explain()

• 실행결과

  

아주 작은 테이블 사이의 조인

• 아주 작은 테이블 사이의 조인을 할 때는 스파크가 조인을 결정하도록 내버려두는 것이 제일 좋다.

References