Spark 쿼리 튜닝

1단계: 스캔(Scan) 확인
  → 읽은 데이터량이 예상보다 크면 → 클러스터링/프루닝 부족

2단계: 셔플(Exchange) 확인
  → Shuffle Write/Read가 크면 → 조인 전략 변경 필요

3단계: 스필(Spill) 확인
  → Spill to Disk 발생하면 → 메모리 부족, 인스턴스 업그레이드

4단계: 스큐(Skew) 확인
  → 특정 태스크만 오래 걸리면 → 데이터 편향 해결 필요

-- 방법 1: 힌트 사용 (한쪽 테이블이 작을 때)
SELECT /*+ BROADCAST(dim_product) */
  f.order_id,
  f.revenue,
  d.product_name,
  d.category
FROM fact_orders f
JOIN dim_product d ON f.product_id = d.product_id;

-- 방법 2: 자동 Broadcast 임계값 조정 (기본 10MB)
SET spark.sql.autoBroadcastJoinThreshold = 100m;  -- 100MB까지 자동 Broadcast

# Python에서 Broadcast 힌트
from pyspark.sql.functions import broadcast

result = (
    fact_orders
    .join(broadcast(dim_product), "product_id")
    .select("order_id", "revenue", "product_name")
)

-- AQE(Adaptive Query Execution) 기반 자동 Skew 처리
-- Databricks에서는 기본 활성화 (spark.sql.adaptive.enabled = true)
-- 추가 설정으로 민감도 조정 가능

SET spark.sql.adaptive.skewJoin.enabled = true;            -- 기본 true
SET spark.sql.adaptive.skewJoin.skewedPartitionFactor = 5; -- 중앙값 대비 5배 이상이면 skew
SET spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes = 256m;  -- 256MB 이상이면 skew

# Salting 기법 (극심한 Skew에 대한 수동 해결)
from pyspark.sql.functions import concat, lit, rand, floor

num_salt_buckets = 10

# 큰 테이블에 salt 추가
orders_salted = (
    orders
    .withColumn("salt", floor(rand() * num_salt_buckets))
    .withColumn("salted_key", concat("customer_id", lit("_"), "salt"))
)

# 작은 테이블을 salt 수만큼 복제
customers_exploded = (
    customers
    .crossJoin(spark.range(num_salt_buckets).withColumnRenamed("id", "salt"))
    .withColumn("salted_key", concat("customer_id", lit("_"), "salt"))
)

# Salted 조인 (Skew 분산)
result = orders_salted.join(customers_exploded, "salted_key")

# 해결 방법 1: 메모리 증가
# Memory-optimized 인스턴스로 전환 (r5, r6i 계열)

# 해결 방법 2: 파티션 수 증가 (데이터 분산)
spark.conf.set("spark.sql.shuffle.partitions", "400")  # 기본 200 → 400

# 해결 방법 3: 조인 전략 변경
# Sort Merge Join 대신 Broadcast Join 사용 (가능한 경우)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "200m")

-- Photon이 사용되고 있는지 확인
-- Query Profile에서 "PhotonGroupingScan", "PhotonShuffleExchange" 등의
-- Photon 연산자가 표시되면 Photon으로 실행 중

-- Photon 활성화 확인
SELECT current_catalog(), current_schema();
-- 클러스터 설정에서 runtime_engine = PHOTON 확인

-- AQE 상세 설정 (Databricks에서는 대부분 기본 활성화)
SET spark.sql.adaptive.enabled = true;                    -- AQE 활성화 (기본)
SET spark.sql.adaptive.coalescePartitions.enabled = true; -- 파티션 병합 (기본)
SET spark.sql.adaptive.localShuffleReader.enabled = true; -- 로컬 셔플 (기본)
SET spark.sql.adaptive.skewJoin.enabled = true;           -- Skew 처리 (기본)

-- 파티션 크기 조정 (기본 64MB)
SET spark.sql.adaptive.advisoryPartitionSizeInBytes = 128m;  -- 대용량 데이터에서 128MB

-- 파티션 병합 최소 크기
SET spark.sql.adaptive.coalescePartitions.minPartitionSize = 1m;  -- 1MB 미만 병합

[AQE 없이 수동 튜닝]
  spark.sql.shuffle.partitions = 200 (기본값)
  → 데이터가 1TB면 파티션당 5GB → 메모리 부족 (Spill)
  → 데이터가 100MB면 파티션당 0.5MB → 과다 파티션 (오버헤드)
  → 수동으로 워크로드마다 조정 필요

[AQE 활성화]
  spark.sql.shuffle.partitions = auto
  → 런타임에 실제 데이터 크기를 보고 자동 조정
  → advisoryPartitionSizeInBytes 기준으로 병합/분할
  → 수동 튜닝 불필요

# 방법 1: 조인 키의 분포 확인
key_distribution = (
    orders
    .groupBy("customer_id")
    .count()
    .orderBy("count", ascending=False)
)

# 상위 10개 키 확인
key_distribution.show(10)
# customer_id | count
# NULL        | 5,000,000  ← NULL 키가 대량! (Skew 원인)
# VIP_001     | 2,000,000  ← 대형 고객
# VIP_002     | 1,500,000

# 방법 2: Query Profile에서 확인
# Task Duration 분포를 보면 특정 태스크만 10x 이상 오래 걸림

# NULL 키 사전 필터링 (가장 간단하고 효과적)
# NULL 키는 조인에 매칭되지 않으므로, 사전에 제거
orders_filtered = orders.filter("customer_id IS NOT NULL")
result = orders_filtered.join(customers, "customer_id")

# 분리 처리 패턴 (NULL도 결과에 필요한 경우)
orders_with_customer = (
    orders.filter("customer_id IS NOT NULL")
    .join(customers, "customer_id")
)
orders_without_customer = (
    orders.filter("customer_id IS NULL")
    .withColumn("customer_name", lit("Unknown"))
)
result = orders_with_customer.union(orders_without_customer)

# Repartition: 데이터를 특정 키로 균등 분배
# 조인 전에 양쪽 테이블을 동일 키로 repartition하면 Shuffle 감소

# 적합한 경우: 동일 키로 여러 번 조인하는 경우
orders_repartitioned = orders.repartition(200, "customer_id")
customers_repartitioned = customers.repartition(200, "customer_id")

# 첫 번째 조인
result1 = orders_repartitioned.join(customers_repartitioned, "customer_id")
# 두 번째 조인 (이미 같은 키로 파티셔닝되어 있으므로 Shuffle 없음)
result2 = orders_repartitioned.join(another_table, "customer_id")

-- 기본 임계값: 10MB
-- Databricks에서는 자동 Broadcast를 위한 임계값을 조정할 수 있음

-- 보수적 설정 (메모리가 제한적인 경우)
SET spark.sql.autoBroadcastJoinThreshold = 10m;

-- 적극적 설정 (메모리 여유가 있는 경우, 권장)
SET spark.sql.autoBroadcastJoinThreshold = 100m;

-- 매우 적극적 설정 (Memory-optimized 인스턴스)
SET spark.sql.autoBroadcastJoinThreshold = 500m;

-- Broadcast 비활성화 (특수한 경우, 예: OOM 발생 시)
SET spark.sql.autoBroadcastJoinThreshold = -1;

-- EXPLAIN으로 실행 계획 확인
EXPLAIN FORMATTED
SELECT /*+ BROADCAST(d) */
  f.order_id, f.revenue, d.product_name
FROM fact_orders f
JOIN dim_product d ON f.product_id = d.product_id
WHERE f.order_date >= '2026-01-01';

[실행 계획 읽기 (아래에서 위로)]

== Physical Plan ==
*(3) Project [order_id, revenue, product_name]        ← 5. 최종 컬럼 선택
+- *(3) BroadcastHashJoin [product_id], [product_id]  ← 4. Broadcast Join 실행
   :- *(3) Filter (order_date >= 2026-01-01)          ← 3. 필터 적용
   :  +- *(3) ColumnarToRow                           ← 2. Photon → Row 변환
   :     +- PhotonScan [order_id, revenue, ...]       ← 1. 팩트 테이블 스캔
   +- BroadcastExchange                               ← Broadcast 배포
      +- PhotonScan dim_product [product_id, ...]     ← 디멘션 테이블 스캔

확인 포인트:
✅ BroadcastHashJoin 사용됨 (SortMergeJoin이 아님)
✅ PhotonScan 사용됨 (Photon 엔진 활성)
✅ Filter가 Scan에 가깝게 위치 (일찍 필터링 = 효율적)

-- ❌ 문제: 불필요한 컬럼까지 모두 읽기 (넓은 테이블에서 치명적)
SELECT * FROM catalog.schema.wide_table WHERE date = '2026-03-01';

-- ✅ 해결: 필요한 컬럼만 선택
SELECT order_id, customer_id, amount
FROM catalog.schema.wide_table WHERE date = '2026-03-01';
-- 100+ 컬럼 테이블에서 3개 컬럼만 선택하면 97% I/O 절감

-- ❌ 문제: 큰 테이블끼리 먼저 조인
SELECT *
FROM fact_orders f           -- 10억 행
JOIN fact_returns r ON f.order_id = r.order_id  -- 1억 행
JOIN dim_product d ON f.product_id = d.product_id;  -- 10만 행

-- ✅ 해결: 작은 테이블(디멘션)을 먼저 조인
SELECT /*+ BROADCAST(d) */ *
FROM fact_orders f
JOIN dim_product d ON f.product_id = d.product_id  -- Broadcast (10만 행)
JOIN fact_returns r ON f.order_id = r.order_id;

-- ❌ 문제: 조인 키의 데이터 타입이 다름 (암묵적 캐스팅 → 인덱스 무효)
SELECT * FROM orders o
JOIN customers c ON o.customer_id = c.customer_id;
-- orders.customer_id: STRING, customers.customer_id: INT
-- → 암묵적으로 CAST 발생 → Data Skipping 무효화

-- ✅ 해결: 명시적 타입 일치
SELECT * FROM orders o
JOIN customers c ON CAST(o.customer_id AS INT) = c.customer_id;
-- 또는 소스에서 타입을 통일

-- ❌ 문제: 고카디널리티 컬럼으로 파티셔닝 (100만+ 파티션)
-- CREATE TABLE ... PARTITIONED BY (user_id)  -- 100만 유저 → 100만 파티션!

-- ✅ 해결: Liquid Clustering으로 전환
ALTER TABLE catalog.schema.user_events
CLUSTER BY (event_date, user_id);
-- 파티셔닝 없이도 효과적인 Data Skipping

# ❌ 문제: 대용량 DataFrame을 Driver로 수집 → OOM
df = spark.table("catalog.schema.big_table")  # 10GB
pandas_df = df.toPandas()  # Driver 메모리 부족!

# ✅ 해결 1: 필터/집계 후 수집
summary = (
    df.filter("date >= '2026-03-01'")
    .groupBy("region")
    .agg({"revenue": "sum"})
)
pandas_df = summary.toPandas()  # 작은 결과만 수집

# ✅ 해결 2: Spark에서 처리 완료 후 저장
df.filter("date >= '2026-03-01'").write.mode("overwrite").saveAsTable("result_table")