Skip to content

PySpark groupBy와 집계: 대규모에서도 정확한 요약 만들기

Updated on

집계가 중복 집계되거나, null 그룹을 건너뛰거나, 카디널리티 문제를 숨기면 리포팅 결과가 망가집니다. PySpark의 groupBy와 agg는 올바른 함수와 별칭을 선택했을 때만 롤업을 정확하게 유지해 줍니다.

이 가이드는 믿을 수 있는 집계 패턴을 다룹니다: 여러 지표를 한 번에 계산하는 방법, distinct 카운트 옵션, null 그룹 처리, 그리고 후속 활용을 위한 결과 정렬까지 설명합니다.

Python Pandas Dataframe에서 코드 없이 빠르게 Data Visualization을 만들고 싶으신가요?

PyGWalker는 시각화를 지원하는 Python 기반 Exploratory Data Analysis 라이브러리입니다. PyGWalker (opens in a new tab)는 pandas dataframe(및 polars dataframe)을 시각적 탐색을 위한 tableau 대체 UI로 바꿔 주어, Jupyter Notebook 기반 데이터 분석과 시각화 워크플로를 간소화합니다.

PyGWalker for Data visualization (opens in a new tab)

집계 함수 선택지

NeedFunctionNotes
전체 개수( null 포함 )count("*")빠름; 행 개수를 셈
Distinct 개수countDistinct(col)정확하지만 더 무거움
근사 Distinctapprox_count_distinct(col, rsd)오차를 조절 가능; 더 빠름
null 안전한 sum/avgsum, avg기본적으로 null을 무시
한 번에 여러 지표 계산agg + alias컬럼 이름을 안정적으로 유지

준비

from pyspark.sql import SparkSession, functions as F
 
spark = SparkSession.builder.appName("groupby-agg").getOrCreate()
 
sales = spark.createDataFrame(
    [
        ("2025-11-01", "US", "mobile", 120.0, "A123"),
        ("2025-11-01", "US", "web",    80.0,  "A124"),
        ("2025-11-01", "CA", "mobile", None,  "A125"),
        ("2025-11-02", "US", "mobile", 200.0, "A126"),
        ("2025-11-02", None, "web",    50.0,  "A127"),
    ],
    "date STRING, country STRING, channel STRING, revenue DOUBLE, order_id STRING",
)

별칭을 활용한 다중 집계

daily = (
    sales.groupBy("date", "country")
    .agg(
        F.count("*").alias("orders"),
        F.sum("revenue").alias("revenue_sum"),
        F.avg("revenue").alias("revenue_avg"),
    )
)
  • alias를 사용하면 이후 단계에서 컬럼 이름이 안정적으로 유지됩니다.
  • sum/avg는 revenue가 null인 행을 자동으로 무시합니다.

Distinct 및 근사 Distinct 카운트

unique_orders = (
    sales.groupBy("date")
    .agg(
        F.countDistinct("order_id").alias("orders_exact"),
        F.approx_count_distinct("order_id", 0.02).alias("orders_approx"),
    )
)
  • 성능이 중요하고, 작은 오차를 허용할 수 있을 때 approx_count_distinct를 사용합니다.

HAVING과 비슷한 방식으로 집계 결과 필터링

high_rev = daily.where(F.col("revenue_sum") > 150)
  • 집계 이후에 필터링하여 SQL의 HAVING과 같은 동작을 구현합니다.

결과 정렬하기

ranked = daily.orderBy(F.col("revenue_sum").desc(), "date")
  • 집계 이후에 정렬합니다; orderBy를 사용하지 않으면 Spark는 결과 순서를 보장하지 않습니다.

null 그룹 키 처리

null_country_stats = (
    sales.groupBy("country")
    .agg(F.count("*").alias("rows"))
)
  • groupBy는 null 키를 유지합니다. 집계 전에 fillna로 대체할지 여부를 명시적으로 결정해야 합니다.

간단한 교차표를 위한 pivot

pivoted = (
    sales.groupBy("date")
    .pivot("channel", ["mobile", "web"])
    .agg(F.sum("revenue"))
)
  • pivot 값 목록을 제한해 컬럼 수가 폭발적으로 늘어나는 것을 방지합니다.

최소 예제의 엔드 투 엔드 롤업

summary = (
    sales
    .groupBy("date", "country")
    .agg(
        F.count("*").alias("orders"),
        F.sum("revenue").alias("revenue_sum"),
        F.countDistinct("order_id").alias("unique_orders"),
    )
    .orderBy("date", "country")
)

이 예제는 일별 안정적인 지표를 생성하고, country에 대한 null 처리 방식을 그대로 유지하며, 컬럼 이름을 이후 조인이나 내보내기에 바로 사용할 수 있도록 정리합니다.