Auto Loader 주요 옵션

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

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.useNotifications", "true")
    # AWS SQS 기반 알림 (자동 설정)
    .option("cloudFiles.region", "ap-northeast-2")
    .load("s3://bucket/data/")
)

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.inferColumnTypes", "true")    # 타입 자동 추론
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/json/")
    .option("multiLine", "true")                      # 여러 줄에 걸친 JSON
    .option("primitivesAsString", "false")             # 숫자를 숫자 타입으로 유지
    .option("allowUnquotedFieldNames", "true")         # 따옴표 없는 필드명 허용
    .option("allowSingleQuotes", "true")               # 작은따옴표 허용
    .load("s3://bucket/json-data/")
)

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/csv/")
    .option("header", "true")                          # 첫 줄을 헤더로 사용
    .option("delimiter", ",")                          # 구분자
    .option("encoding", "UTF-8")                       # 인코딩
    .option("quote", '"')                              # 따옴표 문자
    .option("escape", "\\")                            # 이스케이프 문자
    .option("nullValue", "NULL")                       # NULL 표현 값
    .option("emptyValue", "")                          # 빈 값 표현
    .option("nanValue", "NaN")                         # NaN 표현
    .option("maxColumns", "100")                       # 최대 컬럼 수
    .option("multiLine", "true")                       # 필드 내 줄바꿈 허용
    .load("s3://bucket/csv-data/")
)

# Parquet (스키마가 파일에 내장되어 있어 옵션이 적습니다)
df_parquet = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "parquet")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/parquet/")
    .load("s3://bucket/parquet-data/")
)

# Avro
df_avro = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "avro")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/avro/")
    .option("avroSchema", avro_schema_json)  # 선택: 명시적 스키마 지정
    .load("s3://bucket/avro-data/")
)

df_xml = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "xml")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/xml/")
    .option("rowTag", "record")              # 레코드를 나타내는 XML 태그
    .option("rootTag", "data")               # 루트 XML 태그
    .load("s3://bucket/xml-data/")
)

df_binary = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "binaryFile")
    .option("pathGlobFilter", "*.pdf")
    .load("s3://bucket/documents/")
)
# 결과: path, modificationTime, length, content(binary) 컬럼

# 스키마 진화 설정 예제
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/orders/")
    .option("cloudFiles.inferColumnTypes", "true")
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    # 특정 컬럼 타입 힌트
    .option("cloudFiles.schemaHints",
            "order_id BIGINT, amount DECIMAL(12,2), order_date TIMESTAMP")
    .load("s3://bucket/orders/")
)

# 체크포인트와 함께 전체 파이프라인 설정
(df.writeStream
    .option("checkpointLocation", "s3://bucket/checkpoints/orders/")
    .option("mergeSchema", "true")    # 스키마 변경 시 자동 병합
    .trigger(availableNow=True)       # 현재 사용 가능한 파일만 처리
    .toTable("catalog.bronze.orders")
)

# 2025년 이후 JSON 파일만 처리
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("pathGlobFilter", "*.json")
    .option("modifiedAfter", "2025-01-01T00:00:00Z")
    .option("recursiveFileLookup", "true")
    .load("s3://bucket/data/")
)

# 성능 최적화 예제
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "parquet")
    .option("cloudFiles.maxFilesPerTrigger", "500")
    .option("cloudFiles.maxBytesPerTrigger", "5g")
    .option("cloudFiles.useIncrementalListing", "true")
    .option("cloudFiles.fetchParallelism", "4")
    .load("s3://bucket/large-dataset/")
)

# 에러 처리가 포함된 프로덕션 설정
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/orders/")
    .option("cloudFiles.inferColumnTypes", "true")
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    # 에러 처리
    .option("rescuedDataColumn", "_rescued_data")
    .option("badRecordsPath", "s3://bucket/bad-records/orders/")
    .option("ignoreCorruptFiles", "true")
    .option("ignoreMissingFiles", "true")
    .load("s3://bucket/raw/orders/")
)

-- SDP(선언적 파이프라인)에서 메타데이터 활용
CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT
    *,
    _metadata.file_path AS _source_file,
    _metadata.file_name AS _source_file_name,
    _metadata.file_size AS _source_file_size,
    _metadata.file_modification_time AS _file_modified_at,
    current_timestamp() AS _ingested_at
FROM STREAM read_files(
    's3://bucket/data/',
    format => 'json',
    inferColumnTypes => true,
    rescuedDataColumn => '_rescued_data'
);