[PySpark] 주요 연산 가이드: Transformation, Action

1. 개요

PySpark의 연산은 크게 Transformation과 Action으로 구분된다. 이 구분을 이해하면 “어디서 성능 비용이 발생하는지, 어떤 코드가 실제 실행을 트리거하는지”를 명확히 알 수 있어 전처리와 학습 데이터 준비 단계에서 시행착오를 줄일 수 있다. 

 

2. Transformation vs Action

2.1 정의

• Transformation: 새로운 DataFrame을 반환하지만 즉시 실행되지 않는다(=lazy). Spark는 lineage(연산 이력)만 쌓는다.
• Action: Spark가 실제 Job을 실행하여 결과를 반환하거나 외부 저장소에 기록한다.

2.2 실행 모델 요약

• Action을 만나기 전까지는 Job이 생성되지 않는다.
• Action을 호출하는 순간 DAG가 Job → Stage(셔플 경계 기준) → Task(파티션 단위)로 분해되어 실행된다.

동일 DataFrame에 Action을 여러 번 호출하면 동일 lineage가 반복 실행된다. 재사용할 중간 결과는 cache()/persist() 후 한 번만 Action으로 물리화하는 습관이 성능 최적화의 첫걸음이다.

 

3. Transformation 연산

아래 연산은 모두 lazy이며, 셔플을 유발하는지 여부(좁은/넓은 변환)를 함께 표시한다.

3.1 select / withColumn / drop (Narrow)

• select: 원하는 컬럼만 추출 → 불필요한 컬럼 제거.
• withColumn: 새로운 컬럼 생성 또는 기존 컬럼 수정.
• drop: 특정 컬럼 제거.

from pyspark.sql.functions import col, when
base = df.select("id","user_id","price","ts")
feat = base.withColumn("bucket", when(col("price")<50,"low").otherwise("high"))

3.2 filter (Narrow)

• 조건에 맞는 행만 남김.
• 조인·집계 전에 먼저 적용해 데이터량 축소.

recent = df.filter(col("ts") >= cutoff)

3.3 join (Wide, Shuffle)

• 두 DataFrame을 특정 키로 결합.
• 대규모 셔플 발생 가능, 작은 테이블은 broadcast()로 최적화.

from pyspark.sql.functions import broadcast
joined = fact.join(broadcast(dim.select("id","name")), fact.dim_id==dim.id, "left")

3.4 groupBy + agg (Wide, Shuffle)

• groupBy(): 특정 컬럼 기준으로 그룹화하여 해당 컬럼이 그룹 키가 되고, 해당 컬럼의 동일한 값을 가진 행들이 하나의 그룹으로 묶임
• agg(): 집계 함수(avg, sum, count, max, min 등)와 함께 사용.

user_id	category	price
1	A	100
2	A	200
3	B	300
4	B	400
5	A	50

from pyspark.sql.functions import avg, sum

agg = df.groupBy("category").agg(
    avg("price").alias("avg_price"),
    sum("price").alias("total_price")
)
agg.show()

category	avg_price	total_price
A	116.67	350
B	350.00	700

 

• 위 예시를 보면, category 컬럼을 groupBy로 묶고
• avg, sum 함수를 사용해서 category 동일 값들의 평균과 합에 대한 컬럼을 생성

3.5 distinct / dropDuplicates (Wide, Shuffle)

• distinct: 전체 행에서 중복 제거.
• dropDuplicates: 특정 컬럼 기준으로 중복 제거.

uniq = df.dropDuplicates(["user_id","date"])

3.6 repartition / coalesce (Shuffle/No-shuffle)

• repartition(N): 셔플을 발생시켜 파티션을 균등하게 재분배.
• coalesce(N): 셔플 없이 파티션 수만 축소(병렬성 저하 주의).

wide = df.repartition(800)
compact = wide.coalesce(100)

3.7 Window functions (Wide, Shuffle 가능)

• 파티션별/정렬 조건별로 전후 맥락 집계.
• 순위, 누적합, 최근 데이터 추출 등에 활용.

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number
w = Window.partitionBy("user_id").orderBy(col("ts").desc())
latest = df.withColumn("rn", row_number().over(w)).filter("rn=1")

 

 

4. Action 연산

Action은 “결과를 모으거나 저장하거나 출력”하는 모든 연산이다.

4.1 결과 반환형 Action

• count(): 전체 행 개수 반환.
• first() / head(): 첫 행 또는 n개의 행 반환.
• take(n): 지정한 개수만 수집.
• collect(): 전체 데이터를 드라이버로 수집(메모리 주의).

cnt   = df.count()
head5 = df.take(5)
one   = df.first()
rows  = df.collect()

4.2 표시/디버깅 Action

• show(): 지정한 개수만 콘솔에 출력(내부적으로 수집 후 포맷팅).

df.show(20, truncate=False)

4.3 저장 Action (Writer API)

• write.save(): 파일 시스템에 저장.
• saveAsTable(): Hive/메타스토어에 테이블로 저장.
• jdbc.save(): DB에 직접 적재.

(df.write.mode("overwrite")
    .option("compression","zstd")
    .parquet("s3://bucket/mart/orders/"))

4.4 toPandas (드라이버 수집)

• DataFrame을 Pandas로 변환.
• 소규모 데이터 샘플 확인용으로만 사용.

small = df.select("a","b").sample(0.01, seed=42).limit(10_000)
pdf = small.toPandas()

Action 호출 횟수를 최소화하고, 동일 DF를 재사용한다면 cache()/persist() 후 한 번만 물리화한다.

 

 

5. Transformation vs Action 치트시트

분류	대표 연산 기능	요약	셔플 가능성
Transformation	select	컬럼 선택	Narrow
	withColumn	컬럼 생성/수정	Narrow
	drop	컬럼 제거	Narrow
	filter	조건 행 필터링	Narrow
	join	키 기반 결합	Wide
	groupBy+agg	그룹 집계	Wide
	distinct, dropDuplicates	중복 제거	Wide
	repartition	파티션 재분배	Shuffle
	coalesce	파티션 축소	No-shuffle
	window	파티션/정렬 기반 집계	Wide
Action	count	전체 행 수 반환	실행 트리거
	first, head, take	일부 행 반환	실행 트리거
	collect	전체 데이터 수집	실행 트리거
	show	행 출력(디버깅용)	실행 트리거
	write.save, saveAsTable, jdbc.save	저장	실행 트리거
	toPandas	Pandas 변환	실행 트리거