1
# [producer] Apache Spark는 _SUCCESS 파일을 자동 생성
2
dataset = spark.read.schema("...").json(input_path)
3
dataset.write.mode("overwrite").format("parquet").save(output_path)
4
# → output_path/_SUCCESS 파일이 자동 생성됨
5

6
# [consumer] Airflow FileSensor로 _SUCCESS 대기
7
from airflow.sensors.filesystem import FileSensor
8

9
wait_for_data = FileSensor(
10
    task_id="wait_for_data",
11
    filepath=f"{input_data_path}/_SUCCESS",
12
    mode="reschedule",  # 핵심: 워커 슬롯을 점유하지 않음
13
)

1
[poke 모드]
2
워커 슬롯 점유: ████████████████████ (파일 존재할 때까지 계속)
3
→ 다른 태스크가 실행 불가
4

5
[reschedule 모드]
6
워커 슬롯 점유: ██░░░░██░░░░██░░░░██ (확인 후 반환, 주기적 재확인)
7
→ 사이사이 다른 태스크 실행 가능

1
# [생산자 측] Airflow 태스크로 Readiness Marker 직접 생성
2
from airflow.decorators import task
3
import shutil
4

5
@task
6
def delete_dataset():
7
    shutil.rmtree(dataset_dir, ignore_errors=True)
8

9
@task
10
def generate_dataset():
11
    # 데이터 처리 로직 (생략)
12
    pass
13

14
@task
15
def create_readiness_file():
16
    """반드시 파이프라인의 마지막 태스크로 실행"""
17
    with open(f"{dataset_dir}/COMPLETED", "w") as f:
18
        f.write("")
19

20
# 핵심: readiness marker는 항상 마지막
21
delete_dataset() >> generate_dataset() >> create_readiness_file()

1
# 규약: "다음 파티션이 존재하면 현재 파티션은 완성된 것"
2
#
3
# 잡이 매시간 실행:
4
#   10:00 실행 → partition=10 생성
5
#   11:00 실행 → partition=11 생성
6
#
7
# 소비자가 partition=10을 처리하려면?
8
# → partition=11이 존재하는지 확인
9
# → 존재하면 partition=10은 완성된 것으로 간주
10

11
from airflow.sensors.filesystem import FileSensor
12

13
# partition=11의 존재로 partition=10의 완성을 판단
14
wait_for_next_partition = FileSensor(
15
    task_id="wait_for_partition_readiness",
16
    filepath="/data/output/partition=11",
17
    mode="reschedule",
18
)
19

20
process_partition_10 = SparkKubernetesOperator(
21
    task_id="process_partition_10",
22
    # ...
23
)
24

25
wait_for_next_partition >> process_partition_10

1
[시간축]
2
10:00 파티션 완성 → 소비자가 처리 완료
3
11:00 파티션 완성
4
   ...
5
12:30 — 10:00 시간대의 늦은 이벤트 도착!
6
   → 10:00 파티션에 데이터 추가
7
   → 소비자는 이미 처리 완료 → 누락 발생

1
# 전략 1: 파티션 불변 원칙
2
# 한번 닫힌 파티션은 절대 수정하지 않음
3
# 늦은 데이터는 별도 파티션(예: late_data/)에 적재
4

5
# 전략 2: 변경 통지
6
# 파티션 업데이트 시 소비자에게 알림
7
# 소비자가 해당 파티션을 재처리할지 결정

1
[스케줄 기반의 문제]
2
매일 실행 → 월~목 중 1회만 새 데이터 존재
3
→ 나머지 실행은 "변경 없음" 확인만 하고 종료
4
→ 컴퓨팅 리소스 낭비 + 불필요한 운영 부담

1
[Pull 기반 Trigger]
2
장기 실행 프로세스가 주기적으로 새 이벤트 확인
3
→ 대부분의 시간을 "확인만 하고 0건" 상태로 대기
4
→ 리소스 낭비, 하지만 구현은 단순
5

6
[Push 기반 Trigger] — 권장
7
이벤트 소스가 엔드포인트에 알림을 보냄
8
→ 이벤트 있을 때만 핸들러 인스턴스 생성
9
→ 리소스 효율적, 서버리스(Lambda 등)에 적합

1
# AWS Lambda의 인프라 레벨 에러 관리
2
# 1) 실패한 이벤트를 Dead Letter Queue(DLQ)로 전송
3
# 2) 재시도 횟수 설정
4
# 3) 배치 크기 및 동시성 제어
5

6
# Lambda 설정 (개념적 표현)
7
lambda_config = {
8
    "dead_letter_queue": "arn:aws:sqs:...:trigger-dlq",
9
    "retry_attempts": 2,
10
    "max_concurrency": 10,
11
}
12

13
# DLQ에 쌓인 실패 이벤트를 모니터링하고 수동/자동 재처리