Apache Spark — 느린 잡의 답은 실행 계획과 이벤트 로그에 있다 · DataFrame 과 SQL · 이론

SQL 로 쓰든 DataFrame 으로 쓰든 같은 계획이 되고, 윈도는 행을 줄이지 않는다

한 줄 요약

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

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

왜 두 가지 입구가 있나

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

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

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

어떻게 동작하나

SQL 로 DataFrame 을 부르려면 이름이 필요하다. [시작하기 문서](https://spark.apache.org/docs/4.2.0/sql-getting-started.html#running-sql-queries-programmatically)는 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 문서의 예](https://spark.apache.org/docs/4.2.0/sql-ref-syntax-qry-explain.html#examples)를 보면 아래에서부터 HashAggregate(... partial_sum ...) → Exchange hashpartitioning(k, 200) → HashAggregate(... sum ...) 순서다. 각 파티션이 먼저 자기 몫을 부분 합으로 줄이고, 키로 셔플한 뒤, 모인 부분 합을 다시 더한다. 셔플로 보내는 것은 원래 행이 아니라 키마다 한 줄로 줄인 부분 합이다. 그래서 합계·개수 같은 집계는 자료가 커도 셔플이 생각보다 작다. 예의 200 은 셔플 파티션 수이고, 셔플을 다루는 모듈에서 따로 본다.

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

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

순위 함수의 차이는 동점에서 난다. [rank 문서](https://spark.apache.org/docs/4.2.0/api/python/reference/pyspark.sql/api/pyspark.sql.functions.rank.html)의 설명대로 dense_rank 는 동점 뒤에 빈 순위를 남기지 않고, rank 는 셋이 공동 2등이면 그다음을 5등으로 매긴다. "상위 3" 을 rank <= 3 으로 거르면 동점 때문에 네 줄 이상이 나올 수 있고, row_number 로 거르면 동점 중 누가 들어갈지가 정렬 순서에 맡겨진다. 어느 쪽이 맞는지는 업무가 정한다.

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

정확한 수와 근사한 수

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

현장에서 만나는 모습

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

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

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

실무에서 진짜 중요한 것

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

다음 실습에서 할 것

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