실습 1: 사전 준비와 CSV 수집

-- Unity Catalog에 실습용 카탈로그와 스키마 생성
-- 이미 존재하는 카탈로그가 있다면 해당 카탈로그를 사용해도 됩니다
CREATE CATALOG IF NOT EXISTS training;
USE CATALOG training;

CREATE SCHEMA IF NOT EXISTS auto_loader_lab;
USE SCHEMA auto_loader_lab;

-- 생성 결과 확인
SHOW SCHEMAS IN training;

-- 데이터 저장용 볼륨 생성
CREATE VOLUME IF NOT EXISTS training.auto_loader_lab.raw_data;

-- 볼륨 경로 확인
-- 실제 경로: /Volumes/training/auto_loader_lab/raw_data/
SHOW VOLUMES IN training.auto_loader_lab;

# 노트북 셀에서 실행 - CSV 주문 데이터 생성
import json
from datetime import datetime, timedelta
import random

random.seed(42)  # 재현성을 위해 시드 고정

# --- CSV 주문 데이터 생성 (Batch 1: 100건) ---
csv_header = "order_id,customer_id,product_name,amount,order_date,status\n"
csv_rows = []
for i in range(1, 101):
    date = (datetime(2025, 3, 1) + timedelta(days=random.randint(0, 30))).strftime("%Y-%m-%d")
    status = random.choice(["COMPLETED", "PENDING", "CANCELLED"])
    product = random.choice(["노트북", "키보드", "마우스", "모니터", "헤드셋"])
    amount = round(random.uniform(10000, 500000), 2)
    csv_rows.append(f"{i},{random.randint(1000,9999)},{product},{amount},{date},{status}")

csv_content = csv_header + "\n".join(csv_rows)

# 볼륨에 CSV 파일 저장
dbutils.fs.put(
    "/Volumes/training/auto_loader_lab/raw_data/csv/orders/orders_batch1.csv",
    csv_content,
    overwrite=True
)

print(f"CSV 배치 1 생성 완료: {len(csv_rows)}건")
print("경로: /Volumes/training/auto_loader_lab/raw_data/csv/orders/orders_batch1.csv")

CSV 배치 1 생성 완료: 100건
경로: /Volumes/training/auto_loader_lab/raw_data/csv/orders/orders_batch1.csv

from pyspark.sql.functions import current_timestamp, col

# 경로 설정
source_path      = "/Volumes/training/auto_loader_lab/raw_data/csv/orders/"
checkpoint_path  = "/Volumes/training/auto_loader_lab/checkpoints/csv_orders/"
schema_path      = "/Volumes/training/auto_loader_lab/schema/csv_orders/"

# Auto Loader로 CSV 파일 스트리밍 읽기
df_orders = (
    spark.readStream
    .format("cloudFiles")                               # Auto Loader 포맷
    .option("cloudFiles.format", "csv")                 # 소스 파일 포맷
    .option("cloudFiles.schemaLocation", schema_path)   # 추론된 스키마 저장 위치
    .option("cloudFiles.inferColumnTypes", "true")      # 컬럼 타입 자동 추론 (기본값: 모두 STRING)
    .option("header", "true")                           # CSV 첫 줄을 헤더로 사용
    .option("delimiter", ",")                           # 구분자
    .option("encoding", "UTF-8")                        # 인코딩
    .option("rescuedDataColumn", "_rescued_data")       # 파싱 실패 데이터 보존
    .load(source_path)
)

# _metadata 컬럼으로 파일 출처 정보 추가 (데이터 리니지 추적)
df_with_metadata = (
    df_orders
    .withColumn("_source_file",      col("_metadata.file_path"))
    .withColumn("_file_modified_at", col("_metadata.file_modification_time"))
    .withColumn("_ingested_at",      current_timestamp())
)

# 스키마 확인 (스트리밍이므로 실제 데이터를 읽지 않고 스키마만 출력)
df_with_metadata.printSchema()

root
 |-- order_id: long (nullable = true)
 |-- customer_id: long (nullable = true)
 |-- product_name: string (nullable = true)
 |-- amount: double (nullable = true)
 |-- order_date: date (nullable = true)
 |-- status: string (nullable = true)
 |-- _rescued_data: string (nullable = true)
 |-- _source_file: string (nullable = true)
 |-- _file_modified_at: timestamp (nullable = true)
 |-- _ingested_at: timestamp (nullable = true)

# Bronze 테이블에 스트리밍 쓰기
# trigger(availableNow=True): 현재 도착한 파일을 모두 처리하고 스트림 종료
query = (
    df_with_metadata
    .writeStream
    .option("checkpointLocation", checkpoint_path)
    .option("mergeSchema", "true")         # 스키마 변경 시 자동으로 테이블 스키마 확장
    .trigger(availableNow=True)            # 배치 스타일 실행 (처리 후 자동 종료)
    .toTable("training.auto_loader_lab.bronze_orders")
)

# 스트림 완료까지 대기
query.awaitTermination()
print("수집 완료!")

-- 수집된 데이터 확인
SELECT * FROM training.auto_loader_lab.bronze_orders LIMIT 5;

-- 수집 통계 - 핵심 품질 지표
SELECT
    COUNT(*)                                                   AS total_rows,
    COUNT(DISTINCT _source_file)                               AS source_files,
    MIN(_ingested_at)                                          AS first_ingested,
    MAX(_ingested_at)                                          AS last_ingested,
    SUM(CASE WHEN _rescued_data IS NOT NULL THEN 1 ELSE 0 END) AS rescued_rows,
    SUM(CASE WHEN order_id IS NULL THEN 1 ELSE 0 END)          AS null_order_id
FROM training.auto_loader_lab.bronze_orders;

total_rows | source_files | rescued_rows | null_order_id
-----------+--------------+--------------+--------------
100        | 1            | 0            | 0

# 두 번째 배치 CSV 생성 (order_id 101~200)
csv_rows_2 = []
for i in range(101, 201):
    date = (datetime(2025, 3, 15) + timedelta(days=random.randint(0, 15))).strftime("%Y-%m-%d")
    status = random.choice(["COMPLETED", "PENDING", "CANCELLED"])
    product = random.choice(["노트북", "키보드", "마우스", "모니터", "헤드셋"])
    amount = round(random.uniform(10000, 500000), 2)
    csv_rows_2.append(f"{i},{random.randint(1000,9999)},{product},{amount},{date},{status}")

csv_content_2 = csv_header + "\n".join(csv_rows_2)

dbutils.fs.put(
    "/Volumes/training/auto_loader_lab/raw_data/csv/orders/orders_batch2.csv",
    csv_content_2,
    overwrite=True
)

print(f"배치 2 생성 완료: {len(csv_rows_2)}건")

# 동일한 스트리밍 쿼리를 다시 실행
# → Auto Loader는 체크포인트를 확인하여 batch1은 건너뛰고 batch2만 처리합니다
query2 = (
    df_with_metadata
    .writeStream
    .option("checkpointLocation", checkpoint_path)  # 같은 체크포인트 경로!
    .option("mergeSchema", "true")
    .trigger(availableNow=True)
    .toTable("training.auto_loader_lab.bronze_orders")
)

query2.awaitTermination()
print("증분 처리 완료!")

-- 증분 처리 검증: 행 수가 200으로 늘었는지 확인
SELECT
    COUNT(*)                    AS total_rows,
    COUNT(DISTINCT _source_file) AS source_files,
    MIN(order_id)               AS min_order_id,
    MAX(order_id)               AS max_order_id
FROM training.auto_loader_lab.bronze_orders;
-- 예상 결과: total_rows=200, source_files=2, min=1, max=200

# 의도적으로 잘못된 형식의 CSV 생성
bad_csv = """order_id,customer_id,product_name,amount,order_date,status
201,5001,모니터,150000,2025-04-01,COMPLETED
202,INVALID_ID,키보드,NOT_A_NUMBER,INVALID_DATE,UNKNOWN_STATUS
203,5003,마우스,25000,2025-04-01,COMPLETED
"""

dbutils.fs.put(
    "/Volumes/training/auto_loader_lab/raw_data/csv/orders/orders_batch_bad.csv",
    bad_csv,
    overwrite=True
)

# 스트림 재실행 후 rescued_data 확인
# 타입 불일치 행은 _rescued_data에 원본 JSON으로 보존됩니다

-- 구조적으로 잘못된 행 조회
SELECT order_id, customer_id, amount, _rescued_data
FROM training.auto_loader_lab.bronze_orders
WHERE _rescued_data IS NOT NULL;

df_with_hints = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("cloudFiles.schemaLocation", schema_path)
    .option("cloudFiles.inferColumnTypes", "true")
    # 핵심 컬럼 타입을 명시적으로 고정
    .option("cloudFiles.schemaHints",
            "order_id BIGINT, customer_id BIGINT, amount DECIMAL(12,2), order_date DATE")
    .option("header", "true")
    .option("rescuedDataColumn", "_rescued_data")
    .load(source_path)
)

# 파일명 패턴으로 필터링 (예: orders_*.csv 만 수집)
df_filtered = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("cloudFiles.schemaLocation", schema_path)
    .option("pathGlobFilter", "orders_*.csv")    # glob 패턴
    .option("header", "true")
    .option("rescuedDataColumn", "_rescued_data")
    .load(source_path)
)