Auto Loader란?

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.useNotifications", "false")  # 기본값
    .load("s3://my-bucket/data/")
)

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.useNotifications", "true")
    .load("s3://my-bucket/data/")
)

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.inferColumnTypes", "true")   # 타입까지 추론 (기본: 모두 STRING)
    .option("cloudFiles.schemaLocation", "/checkpoints/orders/schema")  # 스키마 저장 위치
    .load("s3://my-bucket/orders/")
)

.option("cloudFiles.schemaEvolutionMode", "addNewColumns")

.option("cloudFiles.schemaHints",
    "order_id BIGINT, amount DECIMAL(12,2), order_date TIMESTAMP, tags MAP<STRING,STRING>")

.option("rescuedDataColumn", "_rescued_data")

정상 데이터: order_id=1, amount=50000, _rescued_data=NULL
불량 데이터: order_id=NULL, amount=NULL, _rescued_data='{"order_id":"ABC","amount":"invalid"}'

-- read_files() 함수를 사용 (SDP 전용)
CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT
    *,
    _metadata.file_path AS _source_file,
    _metadata.file_name AS _file_name,
    _metadata.file_size AS _file_size,
    _metadata.file_modification_time AS _file_modified_at,
    current_timestamp() AS _ingested_at
FROM STREAM read_files(
    's3://my-bucket/raw/orders/',
    format => 'json',
    inferColumnTypes => true,
    rescuedDataColumn => '_rescued_data'
);

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.inferColumnTypes", "true")
    .option("cloudFiles.schemaLocation", "/checkpoints/orders/schema")
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    .option("rescuedDataColumn", "_rescued_data")
    .load("s3://my-bucket/raw/orders/")
)

# 메타데이터 컬럼 추가
from pyspark.sql.functions import current_timestamp, input_file_name

enriched = (df
    .withColumn("_source_file", input_file_name())
    .withColumn("_ingested_at", current_timestamp())
)

# Delta 테이블로 저장
(enriched.writeStream
    .format("delta")
    .option("checkpointLocation", "/checkpoints/orders/data")
    .option("mergeSchema", "true")
    .outputMode("append")
    .trigger(availableNow=True)  # 현재 사용 가능한 파일만 처리 후 종료
    .toTable("catalog.schema.bronze_orders")
)

[기존 환경: COPY INTO]
- 데이터: 물류 차량 GPS 센서 데이터 (JSON)
- 파일 수: S3 디렉토리에 하루 50만 파일 생성
- 누적: 1년 후 디렉토리에 1.8억 파일

**문제 발생 (COPY INTO)**:
- COPY INTO 실행 시 매번 1.8억 파일 디렉토리를 전체 스캔
- 파일 목록 조회만 45분 소요 (S3 LIST API 한계)
- 실제 처리할 신규 파일은 50만 개인데, 1.8억 개를 확인
- 일일 파이프라인이 4시간 → 점점 길어져 12시간까지 증가
- 간헐적으로 S3 LIST API 타임아웃 → 파이프라인 실패
- 실패 시 수동 재시작 → 새벽 3시 콜 빈발

**Auto Loader 전환 후**:
- 체크포인트에 마지막 처리 위치 기록
- 새 파일 50만 개만 감지하여 처리
- 디렉토리 크기와 무관하게 일정한 성능
- 일일 처리 시간: 35분 (12시간 → 35분)
- 파이프라인 실패: 월 평균 15회 → 0회
- 새벽 3시 콜: 사라짐