Serverless 컴퓨트

-- 서버리스 SQL Warehouse에서 쿼리 실행 (일반 SQL과 동일)
SELECT
    region,
    COUNT(*) AS order_count,
    SUM(revenue) AS total_revenue
FROM production.ecommerce.orders
WHERE order_date >= '2025-01-01'
GROUP BY region
ORDER BY total_revenue DESC;

# 노트북 상단에서 %pip로 라이브러리 설치 가능
%pip install scikit-learn==1.4.0 xgboost==2.0.0

# 설치 후 바로 사용
import sklearn
import xgboost

# databricks.yml - 서버리스 Job 설정
resources:
  jobs:
    daily_etl:
      name: "Daily ETL Pipeline"
      tasks:
        - task_key: "extract"
          notebook_task:
            notebook_path: "/Workspace/etl/extract"
          # environment_key를 지정하면 Serverless로 실행됩니다
          environment_key: "default"

        - task_key: "transform"
          depends_on:
            - task_key: "extract"
          notebook_task:
            notebook_path: "/Workspace/etl/transform"
          environment_key: "default"

        - task_key: "load"
          depends_on:
            - task_key: "transform"
          notebook_task:
            notebook_path: "/Workspace/etl/load"
          environment_key: "default"

      environments:
        - environment_key: "default"
          spec:
            client: "1"
            dependencies:
              - "pandas>=2.0"
              - "requests"
              - "great-expectations"

      schedule:
        quartz_cron_expression: "0 0 6 * * ?"  # 매일 오전 6시
        timezone_id: "Asia/Seoul"

# SDP 파이프라인 노트북
import dlt

@dlt.table(comment="원본 주문 데이터")
def bronze_orders():
    return spark.readStream.format("cloudFiles") \
        .option("cloudFiles.format", "json") \
        .load("/Volumes/production/raw/orders/")

@dlt.table(comment="정제된 주문 데이터")
@dlt.expect_or_drop("valid_amount", "total_amount > 0")
def silver_orders():
    return dlt.read_stream("bronze_orders") \
        .filter("status IS NOT NULL")

-- 서버리스 vs 클래식 비용 비교 쿼리
SELECT
    sku_name,
    billing_origin_product,
    SUM(usage_quantity) AS total_dbus,
    COUNT(DISTINCT usage_date) AS active_days,
    ROUND(SUM(usage_quantity) / COUNT(DISTINCT usage_date), 1) AS avg_daily_dbus
FROM system.billing.usage
WHERE usage_date >= CURRENT_DATE() - INTERVAL 30 DAY
  AND billing_origin_product IN ('JOBS_COMPUTE', 'JOBS_SERVERLESS',
                                  'SQL_COMPUTE', 'SQL_SERVERLESS',
                                  'ALL_PURPOSE_COMPUTE', 'INTERACTIVE_SERVERLESS')
GROUP BY sku_name, billing_origin_product
ORDER BY total_dbus DESC;

-- 현재 클러스터 사용 현황 분석
SELECT
    cluster_name,
    cluster_source,
    COUNT(*) AS job_count,
    SUM(usage_quantity) AS total_dbus
FROM system.billing.usage u
JOIN system.compute.clusters c ON u.usage_metadata.cluster_id = c.cluster_id
WHERE usage_date >= CURRENT_DATE() - INTERVAL 30 DAY
GROUP BY cluster_name, cluster_source
ORDER BY total_dbus DESC;

# Before: Customer-Managed 클러스터 사용
resources:
  jobs:
    my_job:
      tasks:
        - task_key: "etl"
          existing_cluster_id: "0123-456789-abcdef"
          notebook_task:
            notebook_path: "/Workspace/etl/pipeline"

# After: 서버리스로 전환
resources:
  jobs:
    my_job:
      tasks:
        - task_key: "etl"
          # existing_cluster_id 제거
          environment_key: "default"
          notebook_task:
            notebook_path: "/Workspace/etl/pipeline"
      environments:
        - environment_key: "default"
          spec:
            client: "1"
            dependencies:
              - "pandas>=2.0"

-- 마이그레이션 전후 비교
SELECT
    CASE
        WHEN billing_origin_product LIKE '%SERVERLESS%' THEN 'Serverless'
        ELSE 'Classic'
    END AS compute_type,
    SUM(usage_quantity) AS total_dbus,
    COUNT(DISTINCT usage_date) AS active_days
FROM system.billing.usage
WHERE usage_date >= CURRENT_DATE() - INTERVAL 14 DAY
  AND billing_origin_product IN ('JOBS_COMPUTE', 'JOBS_SERVERLESS')
GROUP BY 1;