# 사용자가 지정한 날짜가 필요할 때
from airflow.sdk import get_current_context, task
@task
def process():
ctx = get_current_context()
target = ctx["logical_date"] # 수동 run의 요청 날짜
window = (ctx["data_interval_start"], ctx["data_interval_end"])
# 구간 의미가 필요할 때만 data_interval을 쓴다
3부 · Executor
Celery, Kubernetes, 그리고 Edge
Celery와 Kubernetes는 개념이 2.x와 같고, 워커가 DB 대신 API Server와 통신한다는 점만 다르다. API로만 통신하니 원격지 워커(Edge)도 가능해졌다.
CeleryExecutor
broker에 넣으면 상시 worker가 가져간다
시작 지연이 낮고 자원 경합이 생길 수 있다
k8s에서는 HPA나 KEDA로 늘린다
언제: task가 꾸준히 많고 latency가 중요할 때
KubernetesExecutor
TI마다 pod을 만들고 끝나면 지운다
자원이 격리되는 대신 pod 기동 지연이 있다
pod이 API Server에 닿아야 한다(네트워크, JWT)
언제: 리소스 편차가 크거나 bursty할 때
EdgeExecutor 3.0
다른 네트워크나 지역의 워커가 HTTP로 일을 가져간다
DB 접속 없이 API만 있으면 된다
edge3 provider로 제공된다
언제: 데이터가 있는 원격지에서 실행할 때
3부 · Executor
CeleryKubernetesExecutor 대신 multi-executor를 쓴다
# CeleryKubernetesExecutor는 제거됐다. 여러 executor를 나열하고 task마다 고른다
[core]
executor = CeleryExecutor,KubernetesExecutor # 첫 번째가 기본값
heavy = PythonOperator(
task_id="heavy",
python_callable=run,
executor="KubernetesExecutor",
)
3.2 multi-team(experimental)에서는 팀마다 다른 executor, connection, pool을 쓸 수 있다.
4부 · 알림과 새 기능
끝나지 않은 run도 잡고, 사람을 기다리고, 상태를 기억한다
Callback, Deadline Alerts, HITL, State Store, DB 접근, 운영
4부 · Callback
콜백이 언제, 어디서 도는가
콜백은 워커 실행으로 상태가 바뀔 때만 불린다. CLI나 UI로 상태를 바꾸면 불리지 않는다.
콜백은 리스트로도 지정한다(on_failure_callback=[f1, f2]). 3.2부터 DAG 콜백 context의 TI는 DAG 상태와 관련된 task로 골라진다.
4부 · SLA에서 Deadline Alerts로 (AIP-86)
SLA는 끝난 task만 판정해서 막힌 run을 못 잡았다
SLA 3.0에서 제거
task가 끝나야만 판정한다. 막혀 있는 task는 miss를 못 잡는다
SLA는 task 속성인데 callback은 DAG 속성이다
판정할 때마다 DAG를 다시 파싱한다
Deadline Alerts 3.1부터
기준 시각(큐 진입, logical_date, 고정 시각)에 허용 시간을 더해 판정한다
3.1: triggerer에서 도는 AsyncCallback
3.2: 워커에서 도는 SyncCallback, DAG당 여러 개
3.3: Browse 메뉴에 Deadlines 페이지
4부 · Deadline Alerts
우리 DAG의 sla_miss_callback은 이렇게 옮긴다
from airflow.sdk.definitions.deadline import DeadlineAlert, DeadlineReference, AsyncCallback
with DAG(
"datamodeling_temp_layer_daily_kr",
deadline=DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT, # 큐에 들어간 시각부터
interval=timedelta(hours=2), # 2시간 안에 안 끝나면
callback=AsyncCallback(SlackWebhookNotifier, kwargs={...}),
),
): ...
기존 sla_miss_callback=SLAMissNotifier()의 자리. 3.2 기능은 experimental이라 버전별 문서를 같이 본다.
4부 · 3.1부터 3.3까지의 새 기능
task가 사람을 기다리고, 상태를 기억한다
Human-in-the-Loop 3.1
HITLOperator가 deferred 상태로 멈추고 UI에 입력 폼을 띄운다
권한 있는 사용자가 params와 XCom을 보고 결정한다
3.3: awaiting_input 상태 추가
예: 모델 결과 검수, 적재 전 승인
State Store 3.3
retry와 run을 넘어 남는 key-value
task_state_store: task instance 범위
asset_state_store: asset 범위, 공유 watermark
Spark 재시도 때 기존 job에 다시 붙는다(durable=True)
예: 증분 적재 cursor 저장
그 밖에
3.3 Pluggable Retry Policy: 예외 종류별 재시도와 backoff
3.2 async def를 PythonOperator와 @task에
3.1 trigger rule ALL_DONE_MIN_ONE_SUCCESS
3.3 @result와 /wait API로 run 결과 받기 exp
Pluggable Retry Policy는 공식 블로그에 API 예시가 없어 개념만 적었다.
4부 · State Store
증분 적재 cursor를 run 밖에 남긴다
@task
def extract(**context):
state = context["task_state_store"]
cursor = state.get("last_cursor", default=0) # 지난 run이 남긴 값
rows = fetch_records(since=cursor)
state.set("last_cursor", rows[-1]["id"])
return rows
4부 · DB Connection
외부 DB와 metadata DB는 붙는 길이 다르다
무엇을
어떻게
판정
외부 DB (BigQuery, MySQL 등)
Connection(conn_id) + Hook
허용
Airflow metadata 조회
Airflow Python Client, /api/v2
권장
Airflow metadata 조회
PostgresHook으로 metadata DB 직접
비권장 우회
Airflow metadata 조회
airflow.settings.Session, ORM 모델
차단
콜백은 상태 변화에 반응해 알리는 용도다. 안에서 무거운 query나 대량 insert, update를 하면 장애 지점과 운영 부담이 커진다. dag_run을 BQ로 싱크하는 notifier처럼 metadata를 읽는 코드는 3.x로 옮길 때 점검한다.