SQL で書いても DataFrame で書いても同じ計画になり、ウィンドウ関数は行を減らさない

한국어 원문으로 표시합니다.

한 줄 요약

SQL 문자열과 DataFrame API 는 같은 엔진이 같은 계획으로 바꾸는 두 개의 입구다. 무엇이 더 빠른가가 아니라 무엇이 더 읽기 쉬운가로 고른다. 집계는 행을 묶어 줄이고, 윈도 함수는 행을 그대로 두고 이웃 행을 본다 — 그리고 윈도의 결과는 프레임을 어떻게 잡았느냐에 달려 있다.

概念マップ: 같은 엔진이 같은 계획으로 바꾸는 두 개의 입구・윈도 함수는 행을 그대로 두고 이웃 행을 본다・어느 API 나 언어로 표현했든 같은 실행 엔진을 쓴다.・자기 몫을 부분 합으로 줄이고

왜 두 가지 입구가 있나

데이터 팀에는 SQL 로 생각하는 사람과 파이썬으로 생각하는 사람이 섞여 있다. 그러면 곧 "SQL 로 쓰면 더 빠르다더라" 혹은 "DataFrame 이 최적화가 더 잘 된다더라" 같은 말이 돈다. 둘 다 틀렸다.

Spark SQL 가이드의 첫 문단이 답이다. Spark SQL 은 RDD 보다 자료와 계산의 구조를 더 많이 알고, 그 정보로 최적화를 더 한다. 그리고 결과를 계산할 때는 어느 API 나 언어로 표현했든 같은 실행 엔진을 쓴다. 문서는 이 통일 덕분에 변환마다 더 자연스러운 API 를 골라 오갈 수 있다고 적는다.

같은 엔진이라는 말은 구체적이다. SQL 문자열은 파싱되어 논리 계획이 되고, DataFrame 메서드 호출도 곧바로 논리 계획을 쌓는다. 거기서부터는 같은 분석기·같은 최적화기·같은 물리 계획기를 지난다. 그래서 같은 질문을 두 방식으로 쓰고 explain() 을 찍으면 물리 계획이 같다. 이 실습에서 그것을 직접 견준다.

어떻게 동작하나

SQL 로 DataFrame 을 부르려면 이름이 필요하다. 시작하기 문서는 createOrReplaceTempView 로 DataFrame 을 임시 뷰로 등록하고 spark.sql 로 질의하는 모양을 보여 준다. 같은 쪽은 임시 뷰가 세션 범위라서 만든 세션이 끝나면 사라진다고 적는다. 뷰는 자료를 복사하지 않는다. 계획에 붙인 이름일 뿐이다.

orders.createOrReplaceTempView("orders")
by_sql = spark.sql("""
  SELECT date_trunc('month', order_ts) AS month, sum(qty) AS units
  FROM orders WHERE status = 'paid' GROUP BY 1""")
by_api = (orders.where(F.col("status") == "paid")
          .groupBy(F.date_trunc("month", "order_ts").alias("month"))
          .agg(F.sum("qty").alias("units")))

집계의 물리 계획은 거의 늘 같은 모양이다. EXPLAIN 문서의 예를 보면 아래에서부터 HashAggregate(... partial_sum ...) → Exchange hashpartitioning(k, 200) → HashAggregate(... sum ...) 순서다. 각 파티션이 먼저 자기 몫을 부분 합으로 줄이고, 키로 셔플한 뒤, 모인 부분 합을 다시 더한다. 셔플로 보내는 것은 원래 행이 아니라 키마다 한 줄로 줄인 부분 합이다. 그래서 합계·개수 같은 집계는 자료가 커도 셔플이 생각보다 작다. 예의 200 은 셔플 파티션 수이고, 셔플을 다루는 모듈에서 따로 본다.

윈도 함수 — 행을 줄이지 않는 집계

groupBy 는 분류마다 한 줄을 남긴다. "분류별 상위 3 상품" 은 그렇게 풀 수 없다. 상품 행은 그대로 두고, 같은 분류 안에서 몇 등인지를 옆 칸에 붙여야 한다. 이것이 윈도 함수다. SQL 참조의 윈도 함수 쪽은 순위 함수(RANK·DENSE_RANK·ROW_NUMBER 등)와 분석 함수(LAG·LEAD·FIRST_VALUE 등)를 나누고, 프레임을 ROWS 또는 RANGE 로 적는 문법을 정의한다.

순위 함수의 차이는 동점에서 난다. rank 문서의 설명대로 dense_rank 는 동점 뒤에 빈 순위를 남기지 않고, rank 는 셋이 공동 2등이면 그다음을 5등으로 매긴다. "상위 3" 을 rank <= 3 으로 거르면 동점 때문에 네 줄 이상이 나올 수 있고, row_number 로 거르면 동점 중 누가 들어갈지가 정렬 순서에 맡겨진다. 어느 쪽이 맞는지는 업무가 정한다.

더 조용한 함정은 기본 프레임이다. Window 문서는 정렬이 없으면 파티션 전체(행 기준, 처음부터 끝까지)를, 정렬이 있으면 범위 기준으로 처음부터 현재 행까지 자라는 프레임을 기본으로 쓴다고 적는다. rangeBetween 문서에 따르면 범위 경계는 행의 위치가 아니라 ORDER BY 값을 기준으로 한다. 그래서 날짜로 정렬해 누적합을 내면, 같은 날짜의 행들은 서로를 "현재 행까지" 에 넣어 같은 누적값을 한꺼번에 받는다. 한 줄씩 늘어나는 누적합을 원하면 rowsBetween(Window.unboundedPreceding, Window.currentRow) 로 행 기준 프레임을 적어야 한다.

정확한 수와 근사한 수

고유 고객 수를 정확히 세려면 모든 고객 ID 를 모아 중복을 없애야 하므로 값 자체를 셔플해야 한다. approx_count_distinct는 허용 상대 표준편차 rsd 를 받아 추정하고 기본값은 0.05 다. 문서는 rsd 가 0.01 보다 작아야 한다면 차라리 count_distinct 가 효율적이라고 적는다. 내장 집계 함수 목록은 이 추정이 HyperLogLog++ 로 이뤄진다고 밝힌다. 대시보드의 일간 방문자처럼 5% 오차가 괜찮은 곳에 쓰고, 정산처럼 한 명도 틀리면 안 되는 곳에는 쓰지 않는다.

현장에서 만나는 모습

SQL 문자열의 오류는 실행할 때 난다. 칼럼 이름을 틀리면 파이썬 편집기는 아무 말도 하지 않다가 spark.sql 을 부르는 순간 분석 오류가 난다. DataFrame API 도 칼럼을 문자열로 적으면 마찬가지다. 어느 쪽이든 행동 전에 분석 단계에서 걸리므로, 테스트는 작은 자료로 한 번 끝까지 돌리는 것이 가장 싸다.

두 방식을 섞는 것이 보통이다. 복잡한 윈도와 조인은 SQL 로, 칼럼을 반복해서 만드는 일은 파이썬 반복문과 DataFrame API 로 쓰는 식이다. 계획이 같으니 섞어도 손해가 없다.

임시 뷰는 세션 밖에서 안 보인다. 다른 앱이나 다른 세션에서 같은 이름으로 부르면 테이블을 찾지 못한다. 앱 사이에 자료를 넘기려면 파일이나 테이블로 써야 한다.

실무에서 진짜 중요한 것

SQL 과 DataFrame 은 같은 엔진, 같은 계획이다. 성능이 아니라 읽기 쉬움으로 고르고, 의심되면 explain 으로 견준다.
집계는 부분 집계 → 셔플 → 최종 집계다. 셔플되는 것은 원래 행이 아니라 줄인 부분 결과다.
윈도는 행을 줄이지 않는다. 상위 N 은 순위 함수로 매기고 동점 규칙을 업무에 맞춰 고른다.
누적합은 프레임을 명시한다. 정렬이 있는 윈도의 기본은 범위 프레임이라 같은 값의 행이 한꺼번에 묶인다.
근사 고유 수는 오차가 괜찮은 곳에만 쓴다. 기본 rsd 는 0.05 다.

다음 실습에서 할 것

주문 자료를 임시 뷰로 등록해 SQL 로 월별 매출을 내고, 같은 결과를 DataFrame API 로 다시 쓴 뒤 두 쪽의 계획이 같음을 explain 으로 확인한다. 윈도 함수로 분류별 상위 3 상품을 뽑고, 행 기준 프레임으로 누적합을, lag 로 전월 대비 변화를 계산한다. 마지막으로 고유 고객 수를 근사 함수와 정확한 방법으로 각각 세어 오차를 재고 보고서로 정리한다.