Apache Spark 기초

# DataFrame 생성 예시
df = spark.createDataFrame([
    ("김철수", 28, "서울", 4500000),
    ("이영희", 34, "부산", 5200000),
    ("박민수", 25, "대구", 3800000),
], ["이름", "나이", "도시", "연봉"])

# DataFrame 내용 확인
df.show()
# +------+---+----+-------+
# |  이름|나이|도시|   연봉|
# +------+---+----+-------+
# |김철수| 28|서울|4500000|
# |이영희| 34|부산|5200000|
# |박민수| 25|대구|3800000|
# +------+---+----+-------+

# 이 코드들은 아직 실행되지 않습니다 (계획만 세움)
filtered = df.filter(df.나이 >= 30)           # 30세 이상 필터링
selected = filtered.select("이름", "연봉")      # 필요한 컬럼만 선택
sorted_df = selected.orderBy("연봉", 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
        도시,
        COUNT(*) AS 인원수,
        AVG(연봉) AS 평균연봉
    FROM employees
    GROUP BY 도시
    ORDER BY 평균연봉 DESC
""")

result.show()

%sql
SELECT 도시, COUNT(*) AS 인원수, AVG(연봉) AS 평균연봉
FROM employees
GROUP BY 도시
ORDER BY 평균연봉 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, "대형")
    .when(col("amount") >= 50000, "중형")
    .otherwise("소형")
)

# 4. 집계: 주문 크기별 통계
summary = enriched.groupBy("order_size").agg(
    count("*").alias("주문건수"),
    sum("amount").alias("총매출"),
    avg("amount").alias("평균주문액")
)

# 5. 결과 출력
display(summary)

# 6. 결과를 Delta 테이블로 저장
summary.write.format("delta") \
    .mode("overwrite") \
    .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 강제 (작은 테이블을 명시적으로 브로드캐스트)
result = large_orders.join(
    broadcast(small_stores),  # small_stores를 모든 Executor로 복제
    "store_id"
)

# Broadcast 임계값 조정 (기본 10MB → 100MB)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "104857600")  # 100MB

# 기본값: 200 (대부분의 워크로드에 부적합!)
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.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# Skew Join 감지 임계값 (기본: 256MB, 파티션 중앙값의 5배)
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "268435456")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")

# 셔플 파티션 합치기 목표 크기 (기본: 64MB)
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "134217728")  # 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 디버깅에 유용한 설정
spark.conf.set("spark.sql.adaptive.enabled", "true")  # AQE로 자동 최적화
spark.conf.set("spark.sql.shuffle.partitions", "auto")  # Databricks에서 자동 조정

# 메모리 사용량 모니터링 (Spark UI 외)
print(f"Storage Memory Used: {spark.sparkContext._jsc.sc().getExecutorMemoryStatus()}")