OPTIMIZE와 VACUUM — 심화

-- 전체 테이블 통계 수집
ANALYZE TABLE catalog.schema.orders COMPUTE STATISTICS;

-- 특정 컬럼의 통계만 수집
ANALYZE TABLE catalog.schema.orders
COMPUTE STATISTICS FOR COLUMNS order_date, customer_id, amount;

-- Delta Statistics 수집 (데이터 스킵을 위한 통계)
ANALYZE TABLE catalog.schema.orders
COMPUTE DELTA STATISTICS;

-- 일일 유지 보수 스크립트
-- Step 1: 데이터 압축
OPTIMIZE catalog.schema.orders;

-- Step 2: 불필요한 파일 정리 (30일 보존)
VACUUM catalog.schema.orders RETAIN 720 HOURS;

-- Step 3: 통계 갱신
ANALYZE TABLE catalog.schema.orders COMPUTE STATISTICS;

# PySpark로 여러 테이블 일괄 유지 보수
tables = [
    "catalog.schema.orders",
    "catalog.schema.customers",
    "catalog.schema.products"
]

for table in tables:
    spark.sql(f"OPTIMIZE {table}")
    spark.sql(f"VACUUM {table} RETAIN 720 HOURS")
    spark.sql(f"ANALYZE TABLE {table} COMPUTE STATISTICS")
    print(f"✅ {table} 유지 보수 완료")

상황:
- 20개 주요 테이블, 합계 원본 데이터 약 5TB
- 매일 MERGE(upsert) 작업으로 평균 30%의 행이 업데이트
- MERGE는 기존 파일을 삭제 마킹하고 새 파일을 생성
- 8개월간 이전 파일이 모두 보존

결과:
- S3에 실제 저장된 데이터: 약 18TB (원본의 3.6배)
- 그중 현재 유효한 파일: 5TB
- 삭제 대상(VACUUM으로 정리 가능): 13TB
- 낭비되는 월간 스토리지 비용: 약 $300/월 (연간 $3,600)

-- VACUUM 한 번으로 13TB 정리
VACUUM prod.media.user_events RETAIN 168 HOURS;  -- 7일 보존
-- 삭제된 파일 수: 약 450,000개
-- 해방된 스토리지: 13TB
-- 실행 시간: 약 45분

-- 실수: 보존 기간을 0시간으로 설정 (절대 하면 안 됩니다!)
SET spark.databricks.delta.retentionDurationCheck.enabled = false;
VACUUM prod.crm.customer_master RETAIN 0 HOURS;
-- → 현재 진행 중인 장시간 쿼리가 참조하던 파일까지 삭제
-- → 해당 쿼리가 FileNotFoundException으로 실패
-- → 타임 트래블 완전 불가능

-- Step 1: DRY RUN으로 삭제 대상 파일 수와 크기 확인
VACUUM prod.crm.customer_master DRY RUN;
-- "Found 12,345 files (8.2 GB) that are eligible for deletion"

-- Step 2: 예상 결과가 합리적인지 판단
-- (전체 테이블 크기 대비 삭제 비율이 너무 높으면 주의)

-- Step 3: 트래픽이 적은 시간에 실행
VACUUM prod.crm.customer_master RETAIN 168 HOURS;

-- Step 4: 실행 후 테이블 정상 여부 확인
SELECT COUNT(*) FROM prod.crm.customer_master;
DESCRIBE HISTORY prod.crm.customer_master LIMIT 3;

-- 테이블의 파일 크기 분포 확인
DESCRIBE DETAIL prod.schema.my_table;
-- numFiles, sizeInBytes를 확인
-- 평균 파일 크기 = sizeInBytes / numFiles
-- 평균이 100MB 미만이면 OPTIMIZE 필요

-- 구체적인 파일 크기 분포 (PySpark)

# 파일 크기 분포를 확인하는 진단 스크립트
from delta.tables import DeltaTable

dt = DeltaTable.forName(spark, "prod.schema.my_table")
files_df = dt.detail().select("numFiles", "sizeInBytes").collect()[0]

num_files = files_df["numFiles"]
size_bytes = files_df["sizeInBytes"]
avg_size_mb = (size_bytes / num_files) / (1024 * 1024) if num_files > 0 else 0

print(f"총 파일 수: {num_files:,}")
print(f"총 크기: {size_bytes / (1024**3):.1f} GB")
print(f"평균 파일 크기: {avg_size_mb:.1f} MB")

if avg_size_mb < 100:
    print("⚠️ OPTIMIZE 권장: 평균 파일 크기가 100MB 미만입니다")
elif avg_size_mb < 500:
    print("ℹ️ 양호: 하지만 OPTIMIZE로 추가 개선 가능")
else:
    print("✅ 최적: OPTIMIZE 불필요")

-- Managed Table에서 자동으로 OPTIMIZE와 VACUUM을 실행
-- Unity Catalog에서 카탈로그 레벨로 활성화
ALTER CATALOG my_catalog
SET DBPROPERTIES ('predictive_optimization' = 'ENABLE');

-- 또는 스키마 레벨
ALTER SCHEMA my_catalog.my_schema
SET DBPROPERTIES ('predictive_optimization' = 'ENABLE');

# 모든 테이블을 자동으로 유지보수하는 잡
import datetime

# 유지보수 대상 테이블 목록 (메타데이터에서 동적으로 조회도 가능)
tables = spark.sql("""
    SELECT table_catalog, table_schema, table_name
    FROM system.information_schema.tables
    WHERE table_catalog = 'prod'
      AND table_type = 'MANAGED'
""").collect()

results = []
for t in tables:
    full_name = f"{t.table_catalog}.{t.table_schema}.{t.table_name}"
    try:
        # OPTIMIZE
        opt_result = spark.sql(f"OPTIMIZE {full_name}")
        metrics = opt_result.collect()[0]

        # VACUUM (30일 보존)
        spark.sql(f"VACUUM {full_name} RETAIN 720 HOURS")

        results.append({
            "table": full_name,
            "status": "SUCCESS",
            "files_added": metrics.metrics.numFilesAdded,
            "files_removed": metrics.metrics.numFilesRemoved,
            "timestamp": datetime.datetime.now()
        })
    except Exception as e:
        results.append({
            "table": full_name,
            "status": f"FAILED: {str(e)}",
            "timestamp": datetime.datetime.now()
        })

# 결과를 감사 테이블에 기록
results_df = spark.createDataFrame(results)
results_df.write.mode("append").saveAsTable("ops.maintenance.optimize_vacuum_log")