Airflow

타임존 인식 Dag 만들기

타임존을 인식하는(timezone-aware) Dag을 만드는 것은 간단하다 — 날짜를 다룰 때 `pendulum` 라이브러리로 타임존 정보를 가진 날짜를 사용하기만 하면 된다. Python 표준 라이브러리의 `timezone` 객체는 알려진 한계가 있으므로 사용을 피해야 한다. 즉 Dag의 `start_date`나 스케줄과 관련된 날짜 계산에는 stdlib의 datetime.timezone 대신 pendulum을 쓰는 것이 Airflow가 권장하는 방식이다.

스크립트 파싱과 메타데이터 검증

파이프라인을 테스트할 준비가 되었다면, 먼저 스크립트가 문제없이 파싱되는지 확인해야 한다. ``tutorial.py``를 Dags 폴더에 저장했다면 다음처럼 직접 실행해볼 수 있다.

python ~/airflow/dags/tutorial.py

에러 없이 실행된다면 Dag이 올바르게 설정된 것이다. 다음으로 몇 가지 CLI 명령으로 스크립트를 더 검증해볼 수 있다.

# 데이터베이스 테이블 초기화
airflow db migrate

# 활성 Dag 목록 출력
airflow dags list

# "tutorial" Dag의 태스크 목록 출력
airflow tasks list tutorial

# "tutorial" Dag의 graphviz 표현 출력
airflow dags show tutorial

이 명령들은 실제로 태스크를 실행하지 않고, Dag이 Airflow에 의해 올바르게 인식되고 파싱되는지를 확인하는 용도다.

logical date란 무엇인가

특정 논리적 날짜(logical date)에 대해 개별 태스크 인스턴스를 테스트할 수 있는데, 이는 스케줄러가 특정 날짜·시각에 대해 태스크를 실행하는 상황을 시뮬레이션하는 것이다. 여기서 중요한 점은, 스케줄러는 태스크를 특정 날짜와 시각을 "위해서(for)" 실행하는 것이지, 반드시 그 날짜·시각 "에(at)" 실행하는 것은 아니라는 사실이다.

logical date는 Dag 실행(Dag run)의 이름이 붙는 기준이 되는 타임스탬프다. 일반적으로 워크플로우가 처리하는 시간 구간의 "끝" 시점에 해당하거나, 수동으로 트리거된 경우에는 그 트리거 시점에 해당한다. Airflow는 이 logical date를 이용해 각 실행을 조직화하고 추적한다 — UI, 로그, 코드에서 특정 실행을 가리킬 때 바로 이 값을 기준으로 삼는다. UI나 API로 Dag을 트리거할 때는 직접 logical date를 지정해, 특정 시점을 "기준으로" 워크플로우를 실행시킬 수도 있다.

airflow tasks test vs airflow dags test

다음 명령으로 특정 태스크 인스턴스를 테스트할 수 있다.

# command layout: command subcommand [dag_id] [task_id] [(optional) date]

# print_date 테스트
airflow tasks test tutorial print_date 2015-06-01

# sleep 테스트
airflow tasks test tutorial sleep 2015-06-01

# templated 테스트 — 템플릿이 어떻게 렌더링되는지도 확인 가능
airflow tasks test tutorial templated 2015-06-01

``airflow tasks test``는 태스크 인스턴스를 로컬에서 실행하고 로그를 stdout으로 출력하며, 데이터베이스에 상태를 기록하지 않는다. 개별 태스크 인스턴스를 빠르게 확인해보는 용도로 유용하다.

반면 ``airflow dags test``는 Dag run 하나 전체를 로컬에서 실행한다. ``airflow tasks test``와 달리 실제 Dag run을 생성하고 메타데이터 데이터베이스에 태스크 상태를 기록하므로, 초기화된 데이터베이스와 Airflow가 Dags 폴더에서 정상적으로 직렬화(serialize)할 수 있는 Dag이 필요하다. 이는 전체 Dag을 통째로 테스트할 때 유용하며, 코드에서 프로그래밍적으로 테스트할 때는 ``dag.test()``를 사용할 수 있다.