SQL Warehouse와 파이프라인 최적화

-- 1. Liquid Clustering: 쿼리 패턴에 맞는 데이터 배치
ALTER TABLE catalog.schema.orders
CLUSTER BY (order_date, customer_id);

-- 2. 통계 수집: 옵티마이저에게 정확한 정보 제공
ANALYZE TABLE catalog.schema.orders COMPUTE DELTA STATISTICS;

-- 3. Materialized View: 복잡한 집계 사전 계산
CREATE MATERIALIZED VIEW catalog.schema.mv_daily_orders AS
SELECT order_date, region, COUNT(*) AS order_count, SUM(amount) AS total
FROM catalog.schema.orders
GROUP BY order_date, region;

-- 4. 적절한 데이터 타입 사용
-- ❌ 나쁜 예: 날짜를 문자열로 저장
-- CREATE TABLE ... (order_date STRING, ...)
-- ✅ 좋은 예: 네이티브 날짜 타입 사용 (프루닝/통계 활용)
-- CREATE TABLE ... (order_date DATE, ...)

요청 → Result Cache 확인 → 히트? → 즉시 반환 (0초)
                        ↓ 미스
        Disk Cache 확인 → 히트? → 로컬 SSD에서 읽기 (빠름)
                        ↓ 미스
        리모트 스토리지 (S3/ADLS) 에서 읽기 (느림)

-- Result Cache 확인: 동일 쿼리 재실행 시 Query Profile에서
-- "Result Cache Hit" 표시 여부 확인

-- Disk Cache 상태 확인 (노트북에서)
-- spark.catalog.isCached("table_name")

-- Cache를 최대 활용하는 쿼리 패턴
-- ✅ 동일한 쿼리를 파라미터만 바꿔 실행 (Result Cache 히트율 향상)
-- ✅ 자주 조인하는 디멘션 테이블은 자동으로 Disk Cache에 로드
-- ❌ CURRENT_TIMESTAMP() 등 비결정적 함수 사용 시 Cache 무효화

[Query Profile 읽기 순서]

1. 상단 요약 확인
   - Total Execution Time: 전체 실행 시간
   - Rows Returned: 반환 행 수
   - Data Read: 읽은 데이터량

2. DAG(유향 비순환 그래프) 확인
   - 각 노드(연산)의 실행 시간 비율
   - 가장 넓은(시간 비중 큰) 노드가 병목

3. 상세 지표 확인
   - Rows In vs Rows Out: 필터링 효과
   - Shuffle Bytes: 네트워크 전송량
   - Spill Bytes: 디스크 스필량

[사례 1: 스캔 병목]
Query Profile:
  Scan: 읽은 바이트 = 50GB, 테이블 크기 = 50GB
  Filter: 입력 행 = 10억, 출력 행 = 100만

진단: 50GB 전체를 읽고 1%만 사용 → 프루닝 미작동
해결:
  1. CLUSTER BY (필터 컬럼) 적용
  2. OPTIMIZE 실행
  3. ANALYZE TABLE로 통계 갱신

[사례 2: Shuffle 병목]
Query Profile:
  Exchange (Shuffle Write): 20GB
  Join Type: SortMergeJoin

진단: 양쪽 테이블 모두 Shuffle → 네트워크 비용 높음
해결:
  1. 작은 테이블(< 100MB)이면 BROADCAST 힌트 추가
  2. 조인 키에 Liquid Clustering 적용
  3. 중간 결과를 Materialized View로 사전 계산

[사례 3: Skew 병목]
Query Profile:
  Stage Duration: 최소 2초, 최대 180초, 중앙값 5초

진단: 1개 태스크가 180초 (나머지 대비 36배) → 데이터 편향
해결:
  1. AQE Skew Join 설정 확인 (skewedPartitionFactor 낮추기)
  2. 조인 키의 NULL 값 사전 필터링
  3. 극심한 Skew → Salting 기법 적용

-- 가장 비용이 높은 쿼리 Top 20 (최근 7일)
SELECT
  statement_id,
  executed_by,
  statement_text,
  total_duration_ms,
  rows_produced,
  read_bytes,
  shuffle_write_bytes,
  spill_to_disk_bytes,
  warehouse_id
FROM system.query.history
WHERE start_time >= current_date() - INTERVAL 7 DAYS
ORDER BY total_duration_ms DESC
LIMIT 20;

-- 시간대별 쿼리 패턴 분석
SELECT
  date_trunc('hour', start_time) AS hour,
  COUNT(*) AS query_count,
  AVG(total_duration_ms) / 1000 AS avg_duration_sec,
  SUM(read_bytes) / 1024 / 1024 / 1024 AS total_read_gb
FROM system.query.history
WHERE start_time >= current_date() - INTERVAL 7 DAYS
GROUP BY 1
ORDER BY 1;

-- 동일 쿼리가 반복 실행되는 패턴 (MV 후보)
SELECT
  statement_text,
  COUNT(*) AS execution_count,
  AVG(total_duration_ms) / 1000 AS avg_sec,
  SUM(read_bytes) / 1024 / 1024 / 1024 AS total_read_gb
FROM system.query.history
WHERE start_time >= current_date() - INTERVAL 7 DAYS
AND status = 'FINISHED'
GROUP BY statement_text
HAVING COUNT(*) >= 10  -- 10회 이상 반복
ORDER BY total_read_gb DESC
LIMIT 10;
-- 이 쿼리들은 Materialized View로 전환하면 비용 대폭 절감

-- Spill이 발생한 쿼리 (메모리 부족)
SELECT
  statement_id,
  executed_by,
  LEFT(statement_text, 200) AS query_preview,
  total_duration_ms / 1000 AS duration_sec,
  spill_to_disk_bytes / 1024 / 1024 AS spill_mb
FROM system.query.history
WHERE spill_to_disk_bytes > 0
AND start_time >= current_date() - INTERVAL 7 DAYS
ORDER BY spill_to_disk_bytes DESC
LIMIT 10;

# SDP 파이프라인 성능 설정
import dlt

@dlt.table(
  comment="최적화된 실버 테이블",
  table_properties={
    "pipelines.autoOptimize.managed": "true",      # 자동 최적화
    "pipelines.autoOptimize.zOrderCols": "user_id"  # 자동 Z-ORDER (레거시)
  }
)
def silver_events():
    return (
        dlt.read_stream("bronze_events")
        .withWatermark("event_time", "1 hour")  # 늦은 데이터 처리 한도
        .dropDuplicatesWithinWatermark(["event_id"])  # 중복 제거
    )

# Directory Listing 모드 (기본, 간단한 설정)
df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "/checkpoints/schema")
  .load("s3://bucket/incoming/")
)

# File Notification 모드 (대규모, 실시간)
df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.useNotifications", "true")  # 이벤트 기반
  .option("cloudFiles.schemaLocation", "/checkpoints/schema")
  .load("s3://bucket/incoming/")
)

# 지연시간 최적화 설정 조합
(df.writeStream
  .trigger(processingTime="5 seconds")
  .option("checkpointLocation", "/checkpoints/my_stream")
  .option("maxFilesPerTrigger", "100")        # 트리거당 파일 수 제한
  .option("maxBytesPerTrigger", "50m")         # 트리거당 데이터량 제한
  .outputMode("append")
  .toTable("catalog.schema.realtime_events")
)

# SDP에서 Liquid Clustering 적용
import dlt

@dlt.table(
  comment="클러스터링된 골드 테이블",
  cluster_by=["order_date", "region"],  # Liquid Clustering
  table_properties={
    "quality": "gold"
  }
)
def gold_daily_revenue():
    return (
        dlt.read("silver_orders")
        .groupBy("order_date", "region")
        .agg(
            F.sum("amount").alias("total_revenue"),
            F.countDistinct("customer_id").alias("unique_customers")
        )
    )

# 성능 친화적 Expectation 패턴
@dlt.table
@dlt.expect_or_drop("valid_amount", "amount > 0 AND amount < 1000000")
@dlt.expect("valid_email", "email RLIKE '^[a-zA-Z0-9._%+-]+@'")
def silver_orders():
    return dlt.read_stream("bronze_orders")

# ❌ 성능 비친화적: 외부 테이블 JOIN으로 유효성 검사
# @dlt.expect("valid_product", "product_id IN (SELECT id FROM dim_products)")
# → 매 마이크로배치마다 서브쿼리 실행 → 비용 높음

# ✅ 대안: Broadcast 변수로 유효 목록 캐싱
valid_products = spark.table("dim_products").select("id").collect()
valid_set = {row.id for row in valid_products}
# UDF로 검증

-- SDP 파이프라인 실행 이력 조회
SELECT
  id,
  origin.pipeline_name,
  event_type,
  message,
  timestamp,
  details
FROM event_log(TABLE(pipeline_id))
WHERE event_type IN ('flow_progress', 'planning', 'driver_exception')
ORDER BY timestamp DESC
LIMIT 50;

-- 파이프라인 처리량 모니터링
SELECT
  origin.pipeline_name,
  origin.flow_name,
  details:flow_progress:metrics:num_output_rows AS output_rows,
  details:flow_progress:metrics:num_output_bytes AS output_bytes,
  timestamp
FROM event_log(TABLE(pipeline_id))
WHERE event_type = 'flow_progress'
AND timestamp >= current_date() - INTERVAL 7 DAYS
ORDER BY timestamp DESC;

# 스트리밍 파이프라인 지연시간 모니터링
# Structured Streaming의 진행 상황 확인
query = (
    df.writeStream
    .trigger(processingTime="10 seconds")
    .option("checkpointLocation", "/checkpoints/stream")
    .queryName("my_stream")
    .toTable("catalog.schema.events")
)

# 스트리밍 상태 확인
progress = query.lastProgress
if progress:
    input_rows = progress['numInputRows']
    processing_time = progress['batchDuration']
    print(f"입력 행: {input_rows}, 처리 시간: {processing_time}ms")