파일 감지 모드: Directory Listing vs File Notification

# Directory Listing 모드 (기본값이므로 별도 설정 불필요)
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.useNotifications", "false")  # 기본값
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/")
    .load("s3://bucket/data/")
)

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "parquet")
    .option("cloudFiles.useIncrementalListing", "true")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/")
    .load("s3://bucket/data/year=*/month=*/day=*/")
)

# File Notification 모드 — 자동 프로비저닝
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.useNotifications", "true")
    .option("cloudFiles.region", "ap-northeast-2")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/")
    .load("s3://bucket/data/")
)

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.useNotifications", "true")
    .option("cloudFiles.queueUrl", "https://sqs.ap-northeast-2.amazonaws.com/123456789/my-queue")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/")
    .load("s3://bucket/data/")
)

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetBucketNotification",
                "s3:PutBucketNotification",
                "sns:CreateTopic",
                "sns:DeleteTopic",
                "sns:GetTopicAttributes",
                "sns:Subscribe",
                "sqs:CreateQueue",
                "sqs:DeleteQueue",
                "sqs:GetQueueAttributes",
                "sqs:GetQueueUrl",
                "sqs:ReceiveMessage",
                "sqs:DeleteMessage",
                "sqs:SetQueueAttributes"
            ],
            "Resource": "*"
        }
    ]
}

# 자동 프로비저닝된 리소스 정리
result = spark.read.format("cloudFiles")  \
    .option("cloudFiles.format", "json")  \
    .option("cloudFiles.useNotifications", "true")  \
    .option("cloudFiles.resourceGroup", "auto-loader-rg")  \
    .load("s3://bucket/data/")

# cloudFiles.teardown 명령으로 정리
dbutils.fs.rm("s3://bucket/schema/_notification_resources", True)

프로젝트 초기 (파일 1,000개):
  - Directory Listing 스캔 시간: 5초
  - 트리거 간격: 1분
  - 문제 없음 ✅

3개월 후 (파일 50,000개):
  - Directory Listing 스캔 시간: 3분
  - 트리거 간격: 1분인데 스캔이 3분이라 밀리기 시작
  - "왜 데이터가 늦게 들어오지?" 🤔

6개월 후 (파일 200,000개):
  - Directory Listing 스캔 시간: 15분
  - S3 LIST API 비용: 월 $200+
  - 팀 리더: "이 파이프라인 왜 이렇게 느려?" 😡

1년 후 (파일 500,000개):
  - Directory Listing 스캔 시간: 30분+
  - S3 LIST API 비용: 월 $800+
  - 파이프라인 타임아웃으로 실패 시작 🔥

# 이것을 안 하면 6개월 후에 파이프라인이 느려집니다
# 해결 방법 1: Incremental Listing 활성화
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.useIncrementalListing", "true")  # 핵심!
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/")
    .load("s3://bucket/data/year=*/month=*/day=*/")
)

# 해결 방법 2: File Notification으로 전환
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.useNotifications", "true")
    .option("cloudFiles.region", "ap-northeast-2")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/")
    .load("s3://bucket/data/")
)

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetBucketNotification",
                "s3:PutBucketNotification"
            ],
            "Resource": "arn:aws:s3:::my-data-bucket"
        },
        {
            "Effect": "Allow",
            "Action": [
                "sns:CreateTopic",
                "sns:DeleteTopic",
                "sns:GetTopicAttributes",
                "sns:Subscribe"
            ],
            "Resource": "arn:aws:sns:ap-northeast-2:123456789012:databricks-auto-loader-*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "sqs:CreateQueue",
                "sqs:DeleteQueue",
                "sqs:GetQueueAttributes",
                "sqs:GetQueueUrl",
                "sqs:ReceiveMessage",
                "sqs:DeleteMessage",
                "sqs:SetQueueAttributes"
            ],
            "Resource": "arn:aws:sqs:ap-northeast-2:123456789012:databricks-auto-loader-*"
        }
    ]
}

# 해결 방법: S3 Event Bridge를 사용하면 여러 대상에 이벤트를 라우팅할 수 있습니다
# 또는 기존 SQS 큐를 수동으로 지정하여 충돌을 회피합니다
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.useNotifications", "true")
    .option("cloudFiles.queueUrl",
            "https://sqs.ap-northeast-2.amazonaws.com/123456789/shared-queue")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/")
    .load("s3://bucket/data/")
)

실제 장애 시나리오:
- 금요일 밤: 파이프라인이 OOM으로 실패
- 토요일~일요일: 아무도 모름 (모니터링 알림이 꺼져 있었음)
- 월요일 출근: 파이프라인 재시작
- 화요일: "금요일 밤~일요일 데이터가 없어요!"
- 원인: SQS 보존 기간(4일)은 넘기지 않았지만, DLQ로 빠진 메시지가 있었음

# 전환 전: 기존 체크포인트를 백업해두세요
# 전환 후: 첫 실행 시 File Notification이 구독 시점 이후의 파일만 감지합니다
# → 전환 시점에 이미 존재하는 미처리 파일은 놓칠 수 있습니다!

# 안전한 전환 방법:
# 1. 기존 Directory Listing 스트림을 정상 종료
# 2. File Notification 모드로 새 스트림 시작 (새 체크포인트 경로 사용)
# 3. 전환 시점 전후 데이터 정합성 검증
# 4. 누락된 파일이 있으면 수동으로 backfill