조인 결과는 반드시 행 수로 검증한다 — cache(), left_anti join, BroadcastHashJoin 확인
PySpark Tutorial: Build a Real Pipeline and Read the Spark UI — Darshil Parmar PySpark Tutorial: Build a Real Pipeline and Read the Spark UI — "The join is fast because 400 rows fit in memory" 구간 (14~17페이지)
이 모듈을 다 읽으면
- inner join 전후의 행 수 차이를 확인하지 않으면 고아 레코드가 조용히 사라진다는 것을 설명할 수 있다
- left_anti join으로 매칭되지 않은 레코드만 골라내는 방법을 설명할 수 있다
- cache()가 InMemoryTableScan으로 이어지는 원리와, Spark UI의 SQL 탭에서 BroadcastHashJoin을 직접 확인하는 방법을 설명할 수 있다
3,253건의 정제된 주문을 400건의 고객 차원 테이블과 inner join하면 3,227건만 남아 26건이 사라진다. inner join은 매칭되지 않은 레코드를 아무 경고 없이 버리기 때문에, 조인 전후의 행 수 차이를 항상 확인해야 한다. left_anti join을 쓰면 매칭되지 않은 바로 그 26건만 정확히 골라낼 수 있다. 이 조인 결과에 cache()를 붙인 이유는 DataFrame이 결과가 아니라 레시피이기 때문이며, cache() 없이는 이후의 모든 액션이 두 CSV를 처음부터 다시 읽고 다시 캐스팅하고 다시 조인한다. Spark UI의 SQL 탭을 열어 보면 이 조인이 BroadcastHashJoin으로 실행된 것을 확인할 수 있는데, 이는 400행짜리 고객 테이블이 spark.sql.autoBroadcastJoinThreshold(기본 10MB) 아래이기 때문에 Spark가 자동으로 셔플 없이 전체 사본을 각 태스크에 뿌린 것이다.
3,253건이 들어가 3,227건만 나왔다 — 조인은 항상 행 수로 검증한다
clean.join(customers, on="customer_id", how="inner")를 실행하면 3,253건이었던 clean DataFrame이 3,227건으로 줄어든다. inner join은 왼쪽과 오른쪽 양쪽에 모두 키가 존재하는 행만 남기고 나머지는 아무 경고 없이 버리기 때문에, 조인 전후의 행 수가 다르다는 사실 자체를 반드시 확인해야 한다. 이 26건의 차이를 그냥 넘어가면 실제로 존재하지 않는 고객을 참조하는 주문이 있다는 데이터 품질 문제를 놓치게 된다.
핵심 포인트
- inner join은 매칭되지 않는 행을 조용히 버리므로 조인 전후 행 수 비교가 필수 점검 항목이다
- 3,253건에서 3,227건으로 줄어든 26건의 차이가 곧 존재하지 않는 고객을 참조하는 고아 주문의 개수다
left_anti join으로 사라진 행을 정확히 찾는다
clean.join(customers, on="customer_id", how="left_anti")는 오른쪽 테이블에 매칭되는 키가 없는 왼쪽 행만 정확히 골라낸다. 이 방법으로 확인해 보면 26개의 고아 주문은 서로 다른 26개의 customer_id를 참조하고 있으며, 모두 고객 차원 테이블에 존재하지 않는 ID다. inner join이 조용히 삭제한 바로 그 행들을 left_anti join이 정확히 되짚어 보여주는 것이다.
핵심 포인트
- left_anti join은 오른쪽 테이블에 매칭되지 않는 왼쪽 행만 남긴다
- 26개 고아 주문은 서로 다른 26개의 존재하지 않는 customer_id를 참조하고 있었다
cache()와 InMemoryTableScan, Spark UI SQL 탭에서 확인하는 BroadcastHashJoin
joined = clean.join(customers, on="customer_id", how="inner").cache()에 붙은 cache()는 DataFrame이 결과가 아니라 레시피라는 사실과 관련이 있다. cache() 없이는 이후 호출하는 모든 액션이 두 CSV를 처음부터 다시 읽고, 다시 캐스팅하고, 다시 중복 제거하고, 다시 조인한다. cache()를 걸면 첫 액션 이후 결과가 메모리에 남고, 이후의 모든 실행 계획이 Scan csv 대신 InMemoryTableScan으로 시작하는 것을 Spark UI에서 확인할 수 있다. 다만 세 번 이상 재사용할 DataFrame에만 캐싱을 걸어야 하며, 캐싱한 결과는 unpersist()를 호출하기 전까지 메모리를 계속 차지한다. Spark UI의 SQL/DataFrame 탭에서 이 조인의 실행 계획을 열어 보면 노드 이름이 BroadcastHashJoin으로 표시되는데, 이는 400행짜리 customers 테이블이 spark.sql.autoBroadcastJoinThreshold(기본값 10485760바이트, 10MB — 앞선 모듈에서 정정한 값과 동일하게 재확인됨) 아래이기 때문에 Spark가 셔플 없이 테이블 전체 사본을 모든 태스크에 뿌린 것이다. 팩트 테이블과 작은 차원 테이블을 조인할 때는 이 방식이 사실상 무료에 가깝다.
핵심 포인트
- cache()는 첫 액션 이후 결과를 메모리에 보관해, 세 번 이상 재사용할 DataFrame의 재계산을 막아준다
- 캐싱 여부는 실행 계획이 Scan csv로 시작하는지 InMemoryTableScan으로 시작하는지로 Spark UI에서 확인할 수 있다
- SQL 탭의 BroadcastHashJoin 노드는 작은 테이블이 autoBroadcastJoinThreshold(기본 10MB) 아래일 때 셔플 없이 전체를 복제해 조인했다는 뜻이다