ML/AI 최적화와 성능 진단

# Model Serving 엔드포인트 설정 (최적화 버전)
import mlflow

# 모델 등록 시 최적화 힌트
mlflow.pyfunc.log_model(
    artifact_path="model",
    python_model=my_model,
    pip_requirements=["scikit-learn==1.3.0"],  # 최소 의존성
    metadata={"serving_optimization": "latency"}
)

import mlflow
from pyspark.sql.functions import struct

# Spark UDF로 배치 추론 병렬화
model_uri = "models:/fraud_detection/production"
predict_udf = mlflow.pyfunc.spark_udf(spark, model_uri)

# 대용량 데이터셋에 병렬 추론 적용
predictions = (
    spark.table("catalog.schema.transactions")
    .withColumn("prediction", predict_udf(struct("feature1", "feature2", "feature3")))
)

# 최적화 팁: 파티션 수를 코어 수의 2~3배로 설정
predictions.repartition(spark.sparkContext.defaultParallelism * 2)

# Vector Search 쿼리 최적화
results = index.similarity_search(
    query_text="고객 이탈 예측 방법",
    columns=["content", "source", "page"],  # 필요한 컬럼만 반환
    filters={"category": "ml", "year": 2025},  # 메타데이터 필터로 범위 축소
    num_results=5  # 필요한 만큼만 요청
)

# GPU 사용률 확인 (노트북에서)
import subprocess
result = subprocess.run(['nvidia-smi', '--query-gpu=utilization.gpu,memory.used,memory.total',
                        '--format=csv,noheader,nounits'], capture_output=True, text=True)
print(result.stdout)

# GPU 사용률 지속 모니터링 (학습 중)
# nvidia-smi dmon -s u -d 5  # 5초 간격으로 사용률 출력

from pyspark.ml.torch.distributor import TorchDistributor

def train_fn():
    import torch
    import torch.distributed as dist
    from torch.nn.parallel import DistributedDataParallel as DDP
    
    # 분산 환경 초기화 (TorchDistributor가 자동 설정)
    local_rank = int(os.environ.get("LOCAL_RANK", 0))
    device = torch.device(f"cuda:{local_rank}")
    
    model = MyModel().to(device)
    model = DDP(model, device_ids=[local_rank])
    
    optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)
    
    for epoch in range(10):
        for batch in train_loader:
            loss = model(batch)
            loss.backward()
            optimizer.step()
            optimizer.zero_grad()
    
    return model

# Databricks에서 분산 학습 실행
distributor = TorchDistributor(
    num_processes=4,     # 총 GPU 수
    local_mode=False,    # 멀티 노드
    use_gpu=True
)
model = distributor.run(train_fn)

# Mixed Precision Training (메모리 절감 + 속도 향상)
from torch.cuda.amp import autocast, GradScaler

scaler = GradScaler()

for batch in train_loader:
    optimizer.zero_grad()
    
    with autocast():  # FP16으로 연산
        output = model(batch)
        loss = criterion(output, labels)
    
    scaler.scale(loss).backward()  # 그래디언트 스케일링
    scaler.step(optimizer)
    scaler.update()

# Gradient Accumulation (큰 배치 효과를 작은 GPU로)
accumulation_steps = 4
for i, batch in enumerate(train_loader):
    loss = model(batch) / accumulation_steps
    loss.backward()
    
    if (i + 1) % accumulation_steps == 0:
        optimizer.step()
        optimizer.zero_grad()

# 단일 노드: Pandas + scikit-learn (데이터 < 100GB)
import pandas as pd
from sklearn.ensemble import GradientBoostingClassifier
import mlflow

# Spark 테이블을 Pandas로 변환 (데이터가 메모리에 들어갈 때)
df = spark.table("catalog.schema.features").toPandas()

with mlflow.start_run():
    model = GradientBoostingClassifier(n_estimators=100)
    model.fit(df[features], df[target])
    mlflow.sklearn.log_model(model, "model")

# 분산: Spark ML (데이터 > 100GB)
from pyspark.ml.classification import GBTClassifier
from pyspark.ml.feature import VectorAssembler

assembler = VectorAssembler(inputCols=features, outputCol="features")
gbt = GBTClassifier(featuresCol="features", labelCol="target", maxIter=100)

pipeline = Pipeline(stages=[assembler, gbt])
model = pipeline.fit(spark.table("catalog.schema.features"))

# 대용량 데이터에서 Feature Engineering은 Spark로,
# 학습은 단일 노드로 하는 하이브리드 패턴

# Step 1: Spark로 대규모 Feature Engineering (분산)
features_df = (
    spark.table("catalog.schema.raw_data")  # 1TB
    .withColumn("feature_1", ...)
    .withColumn("feature_2", ...)
    .groupBy("entity_id")
    .agg(...)
)
features_df.write.saveAsTable("catalog.schema.features")  # 50GB

# Step 2: 학습용 샘플링 (50GB → 5GB)
sample_df = features_df.sample(fraction=0.1).toPandas()

# Step 3: 단일 노드에서 학습 (최적화된 라이브러리 활용)
import xgboost as xgb
model = xgb.XGBClassifier(tree_method="gpu_hist", gpu_id=0)
model.fit(sample_df[feature_cols], sample_df[target_col])

# Step 4: 전체 데이터에 Spark UDF로 추론 (분산)
predict_udf = mlflow.pyfunc.spark_udf(spark, "models:/my_model/production")
predictions = features_df.withColumn("prediction", predict_udf(struct(*feature_cols)))

[Driver OOM 체크리스트]
□ collect(), toPandas(), show(매우 큰 수) 사용하고 있는가?
  → 필터/집계 후 소량만 수집
□ Broadcast 변수가 너무 큰가? (> 1GB)
  → Broadcast 임계값 낮추기
□ 대시보드/노트북에서 대량 결과를 표시하는가?
  → LIMIT 추가

[Executor OOM 체크리스트]
□ Spill to Disk가 발생하는가?
  → 파티션 수 증가: spark.sql.shuffle.partitions = 2000
□ 데이터 Skew가 있는가?
  → AQE Skew Join 활성화, Salting 적용
□ 인스턴스 메모리가 부족한가?
  → Memory-optimized 인스턴스 (r5, r6i)

[GPU OOM 체크리스트]
□ 배치 사이즈가 너무 큰가?
  → 배치 사이즈 절반으로 줄이기
□ Mixed Precision을 사용하고 있는가?
  → torch.cuda.amp 적용 (메모리 50% 절감)
□ Gradient Checkpointing을 사용하는가?
  → model.gradient_checkpointing_enable() (메모리 절감)

# Driver OOM 방지: 대량 데이터 수집 대신 파일로 저장
# ❌ 위험
large_df = spark.table("catalog.schema.big_table")
pandas_df = large_df.toPandas()  # OOM!

# ✅ 안전
large_df.write.format("delta").save("/tmp/result")
# 필요한 부분만 읽기
small_df = spark.read.format("delta").load("/tmp/result").filter("date = '2026-03-01'").toPandas()

[GC 문제 증상]
- Spark UI의 Executor 탭에서 GC Time이 전체 시간의 10% 이상
- 태스크 실행 시간이 불규칙 (GC Pause로 인한 간헐적 지연)
- "GC overhead limit exceeded" 에러

# GC 튜닝 설정 (클러스터 Spark 설정)
spark.conf.set("spark.executor.extraJavaOptions",
    "-XX:+UseG1GC -XX:G1HeapRegionSize=16m -XX:InitiatingHeapOccupancyPercent=35")

# GC 로그 활성화 (디버깅용)
spark.conf.set("spark.executor.extraJavaOptions",
    "-XX:+UseG1GC -Xlog:gc*:file=/tmp/gc.log:time,uptime:filecount=5,filesize=10m")

# ❌ GC 부담 높은 패턴: 불필요한 중간 DataFrame 생성
df1 = spark.table("table1")
df2 = df1.filter("date > '2026-01-01'")
df3 = df2.withColumn("new_col", ...)
df4 = df3.groupBy("region").agg(...)
df5 = df4.orderBy("total")

# ✅ GC 친화적 패턴: 체이닝으로 중간 참조 제거
result = (
    spark.table("table1")
    .filter("date > '2026-01-01'")
    .withColumn("new_col", ...)
    .groupBy("region").agg(...)
    .orderBy("total")
)

# ✅ 사용 완료한 DataFrame 캐시 해제
cached_df.unpersist()

[Executor 메모리 구조]

전체 Executor 메모리 (spark.executor.memory)
├── Spark Memory (60%)
│   ├── Storage Memory: 캐시된 데이터, Broadcast
│   └── Execution Memory: Shuffle, Sort, Join, 집계
│
├── User Memory (40%)
│   ├── Python UDF 데이터
│   └── 사용자 데이터 구조
│
└── Reserved Memory (300MB 고정)

Off-Heap Memory (spark.memory.offHeap.size)
├── Photon 엔진 메모리
└── Arrow 직렬화 버퍼

# 현재 메모리 설정 확인
print(f"Executor Memory: {spark.conf.get('spark.executor.memory')}")
print(f"Driver Memory: {spark.conf.get('spark.driver.memory')}")
print(f"Memory Fraction: {spark.conf.get('spark.memory.fraction')}")
print(f"Off-Heap Size: {spark.conf.get('spark.memory.offHeap.size', 'not set')}")

# 캐시된 데이터 크기 확인
for table_name in ["table1", "table2"]:
    if spark.catalog.isCached(table_name):
        print(f"{table_name}: cached")

□ Query Profile에서 가장 시간이 오래 걸리는 단계는?
  → Scan이 오래 걸림 → Step 2로
  → Join이 오래 걸림 → Step 3으로
  → Aggregate가 오래 걸림 → Step 4로

□ 읽은 데이터량 vs 필요한 데이터량 비교
  → 과도한 스캔 → Liquid Clustering 적용했는가?
  → 클러스터링 적용됨 → OPTIMIZE를 최근에 실행했는가?
  → OPTIMIZE 실행됨 → 클러스터링 키가 쿼리 패턴과 맞는가?
□ ANALYZE TABLE로 통계를 수집했는가?
□ 파일 수가 과도하지 않은가? (Small File 문제)

□ 조인 유형 확인 (Broadcast vs SortMerge)
  → 한쪽 테이블이 작은데 SortMerge? → Broadcast 힌트 추가
  → 양쪽 모두 대용량? → Shuffle Partition 수 조정
□ 데이터 편향(Skew) 존재하는가?
  → 특정 태스크만 오래 걸림 → AQE Skew Join 설정 확인
  → 극심한 Skew → Salting 기법 적용
□ 조인 키에 NULL이 많은가?
  → NULL 키는 사전 필터링

□ Shuffle 데이터량이 과도한가?
  → Pre-aggregate 가능한가? (서브쿼리에서 먼저 집계)
  → Shuffle Partition 수 조정 (기본 200, 데이터에 따라 조절)
□ Spill to Disk 발생하는가?
  → 메모리 확대 (Memory-optimized 인스턴스)
  → Partition 수 증가로 단위 데이터 축소
□ Photon이 활성화되어 있는가?
  → SQL 집계 워크로드에 Photon 적용 시 2~8x 향상

□ 클러스터/Warehouse 사이즈가 적절한가?
  → CPU 사용률 > 80% 지속 → 스케일 업
  → 메모리 사용률 > 80% → Memory-optimized 인스턴스
  → I/O 대기 높음 → Storage-optimized 인스턴스
□ Autoscaling이 적절히 동작하는가?
  → Min/Max 워커 수 검토
□ 네트워크 병목이 있는가?
  → 리전 간 데이터 전송 최소화
  → 같은 리전에 컴퓨트와 스토리지 배치