Apache Spark 기초

# DataFrame 생성 예시
df = spark.createDataFrame([
    ("Alice", 28, "Seoul",  4500000),
    ("Bob",   34, "Busan",  5200000),
    ("Carol", 25, "Daegu",  3800000),
], ["name", "age", "city", "salary"])

# DataFrame 내용 확인
df.show()
# +-----+---+-----+-------+
# | name|age| city| salary|
# +-----+---+-----+-------+
# |Alice| 28|Seoul|4500000|
# |  Bob| 34|Busan|5200000|
# |Carol| 25|Daegu|3800000|
# +-----+---+-----+-------+

# 이 코드들은 아직 실행되지 않습니다 (계획만 세움)
filtered  = df.filter(df.age >= 30)                    # 30세 이상 필터링
selected  = filtered.select("name", "salary")          # 필요한 컬럼만 선택
sorted_df = selected.orderBy("salary", ascending=False)  # 연봉 내림차순 정렬

# Action: 이 시점에 위의 모든 변환이 실행됩니다
sorted_df.show()        # 결과를 화면에 출력 (Action!)
sorted_df.count()       # 건수를 반환 (Action!)
sorted_df.collect()     # 전체 데이터를 Driver로 가져옴 (Action!)
sorted_df.write.save()  # 파일로 저장 (Action!)

# DataFrame을 임시 뷰로 등록
df.createOrReplaceTempView("employees")

# SQL로 분석
result = spark.sql("""
    SELECT
        city,
        COUNT(*)   AS headcount,
        AVG(salary) AS avg_salary
    FROM employees
    GROUP BY city
    ORDER BY avg_salary DESC
""")

result.show()

%sql
SELECT city, COUNT(*) AS headcount, AVG(salary) AS avg_salary
FROM employees
GROUP BY city
ORDER BY avg_salary DESC

from pyspark.sql.functions import col, sum, avg, count, when, upper

# 1. 데이터 읽기 (Delta 테이블에서)
orders = spark.read.table("catalog.schema.orders")

# 2. 필터링: 완료된 주문만
completed = orders.filter(col("status") == "COMPLETED")

# 3. 변환: 새 컬럼 추가 (주문 금액 기준으로 사이즈 라벨링)
enriched = completed.withColumn(
    "order_size",
    when(col("amount") >= 100000, "large")
    .when(col("amount") >=  50000, "medium")
    .otherwise("small")
)

# 4. 집계: 주문 크기별 통계
summary = enriched.groupBy("order_size").agg(
    count("*").alias("order_count"),
    sum("amount").alias("total_revenue"),
    avg("amount").alias("avg_order_value"),
)

# 5. 결과 출력
display(summary)

# 6. 결과를 Delta 테이블로 저장
(summary.write
    .format("delta")
    .mode("overwrite")               # 기존 데이터 덮어쓰기 (append/ignore/errorifexists 와 택1)
    .saveAsTable("catalog.schema.order_summary"))

사용자 코드 (SQL/DataFrame API)
    ↓
[1] Unresolved Logical Plan — 파싱된 논리 계획 (테이블/컬럼 미확인)
    ↓ (Catalog에서 테이블/컬럼 확인)
[2] Resolved Logical Plan — 확인된 논리 계획
    ↓ (최적화 규칙 적용)
[3] Optimized Logical Plan — 최적화된 논리 계획
    ↓ (물리 전략 선택)
[4] Physical Plan — 물리 실행 계획
    ↓ (코드 생성)
[5] RDD 연산 — 실제 분산 실행

# 실행 계획 확인 방법
df = spark.sql("""
    SELECT customer_id, SUM(amount)
    FROM catalog.schema.orders
    WHERE order_date >= '2025-01-01'
    GROUP BY customer_id
""")

# 논리 + 물리 실행 계획 출력
df.explain(mode="extended")

# Databricks에서는 Spark UI의 SQL 탭에서 시각적으로 확인 가능

-- 테이블 통계 수집 (CBO 활성화의 핵심)
ANALYZE TABLE catalog.schema.orders COMPUTE STATISTICS;

-- 컬럼 단위 통계 수집 (더 정밀한 최적화)
ANALYZE TABLE catalog.schema.orders
COMPUTE STATISTICS FOR COLUMNS customer_id, order_date, amount;

from pyspark.sql.functions import broadcast

# Broadcast Join 강제 (작은 테이블을 명시적으로 브로드캐스트)
# - broadcast(df) hint 는 임계값을 무시하고 강제로 broadcast 시도
# - 단, 실제 크기가 Driver memory 의 일정 비율을 넘으면 fallback 됨
result = large_orders.join(
    broadcast(small_stores),  # small_stores 전체를 모든 Executor 로 복제
    "store_id"
)

# spark.sql.autoBroadcastJoinThreshold
#   - 의미: 이 크기 미만의 테이블은 자동으로 Broadcast Join 후보로 선택
#   - 기본값: 10MB (10 * 1024 * 1024 = 10485760)
#   - 단위: bytes
#   - 비활성화: -1 (자동 broadcast 사용 안 함)
#   - 튜닝: Driver 메모리가 충분하고 dim 테이블이 큰 경우 100MB ~ 500MB 까지 상향
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(100 * 1024 * 1024))  # 100MB

# spark.sql.shuffle.partitions
#   - 의미: GROUP BY / JOIN / ORDER BY 등 셔플 후 생성될 파티션 수
#   - 기본값: 200 (대부분의 워크로드에 부적합)
#   - 영향: 너무 적으면 OOM/GC, 너무 많으면 스케줄링 오버헤드·소형 파일
#   - Databricks 권장: AQE 활성화 + "auto" 로 두면 런타임에 자동 조정
spark.conf.get("spark.sql.shuffle.partitions")  # "200"

# 데이터 규모에 맞게 조정
# 일반 가이드: 셔플 후 각 파티션이 100~200MB 가 되도록 설정
# 예: 셔플 데이터 100GB → 100GB / 128MB ≈ 800 파티션
spark.conf.set("spark.sql.shuffle.partitions", "800")

# AQE 관련 설정 (Databricks 에서는 기본 활성화)

# spark.sql.adaptive.enabled
#   - AQE 마스터 스위치. true 일 때 아래 세부 기능이 동작
#   - 기본값: true (Databricks) / false (OSS Spark < 3.2)
spark.conf.set("spark.sql.adaptive.enabled", "true")

# spark.sql.adaptive.coalescePartitions.enabled
#   - 셔플 후 작은 파티션을 자동 병합 (Coalescing Post-Shuffle Partitions)
#   - 기본값: true
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

# spark.sql.adaptive.skewJoin.enabled
#   - 편향 파티션을 자동 분할
#   - 기본값: true
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# Skew Join 감지 임계값
#   - skewedPartitionThresholdInBytes: 이 크기 이상이면 skew 후보 (기본 256MB)
#   - skewedPartitionFactor:           파티션 중앙값의 N 배 이상이면 skew (기본 5)
#   - 두 조건을 동시에 만족할 때 분할
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", str(256 * 1024 * 1024))
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")

# spark.sql.adaptive.advisoryPartitionSizeInBytes
#   - 셔플 파티션을 병합/분할할 때 목표 크기
#   - 기본값: 64MB
#   - 튜닝: 클러스터 코어가 많을수록 키워서 task 수 줄임 (128~256MB 권장)
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", str(128 * 1024 * 1024))  # 128MB

Spark UI → Stages 탭 → Task Duration 확인
- 대부분의 Task: 10초
- 하나의 Task: 10분  ← Skew!

-- Databricks에서 Skew Join Hint 사용
SELECT /*+ SKEW('orders', 'region') */
    o.*, s.store_name
FROM orders o
JOIN stores s ON o.region = s.region;

-- 특정 Skew 값을 명시
SELECT /*+ SKEW('orders', 'region', ('서울', '경기')) */
    o.*, s.store_name
FROM orders o
JOIN stores s ON o.region = s.region;

from pyspark.sql.functions import col, concat, lit, floor, rand, explode, array

SALT_BUCKETS = 10

# 큰 테이블: 키에 salt 추가
orders_salted = orders.withColumn(
    "salted_key",
    concat(col("region"), lit("_"), floor(rand() * SALT_BUCKETS).cast("string"))
)

# 작은 테이블: salt 값 전체를 explode로 복제
salt_values = [str(i) for i in range(SALT_BUCKETS)]
stores_exploded = stores.withColumn(
    "salt", explode(array([lit(s) for s in salt_values]))
).withColumn(
    "salted_key", concat(col("region"), lit("_"), col("salt"))
)

# Salted 키로 조인
result = orders_salted.join(stores_exploded, "salted_key")

Spark UI → Stages 탭에서 확인:
- Shuffle Spill (Memory): 메모리에서 처리된 데이터
- Shuffle Spill (Disk): 디스크로 유출된 데이터
- Disk Spill 비율이 높으면 메모리 부족 신호!

# OOM 디버깅에 유용한 설정

# AQE 로 셔플 파티션·skew 자동 보정 → 가장 먼저 켜기
spark.conf.set("spark.sql.adaptive.enabled", "true")

# spark.sql.shuffle.partitions = "auto"
#   - Databricks 전용 값. AQE 가 클러스터 코어/데이터 크기 기반으로 동적 결정
#   - OSS Spark 에서는 정수 값만 허용
spark.conf.set("spark.sql.shuffle.partitions", "auto")

# Executor 별 메모리 사용량 스냅샷
#   - getExecutorMemoryStatus() : {executor_id: (max_memory, remaining_memory)}
#   - private API 이므로 운영 코드보다는 디버깅에서만 사용
status = spark.sparkContext._jsc.sc().getExecutorMemoryStatus()
print(f"Executor memory status: {status}")