## Dependency
```python
from __future__ import annotations
import datetime
import pendulum
from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
```
- `__future__` : 파이썬의 모듈로 향후 파이썬 버전의 기능을 현재에 사용할 수 있도록 하는 모듈이다.[^1]
- `pendulum` : 파이썬의 `datetime` 대체하는 날짜 모듈이다.[^2]
## DAG 설정
```python
with DAG(
dag_id="my_first_dags",
schedule="@once",
start_date=pendulum.datetime(2021, 1, 1, tz="Asia/Seoul"),
catchup=False,
dagrun_timeout=datetime.timedelta(minutes=60),
tags=["example", "example2"],
params={"example_key": "example_value"},
) as dag:
# ...
```
- Airflow의 예시 파일에서는 DAG 인스턴스를 위와 같이 context manager로 선언한다. 그냥 일반 클래스처럼 인스턴스를 생성할 수도 있으나 context manager를 쓰는 편이 더 가시성도 좋고 안정적이다. [^3]
### Parameter
#### dag_id
DAG 이름. Airflow DAGs에 표시되는 이름이 이거다.
#### schedule
DAG가 실행되는 스케줄이다. Preset 방식과 Cron 방식으로 표현이 가능한데, Preset 방식은 `@[Preset 이름]`으로 표현하고, Cron 방식은 `"[분] [시] [일] [월] [요일]"` 으로 표현한다. Airflow 과거 API 문서를 찾아보면 아래와 같이 정리되어 있다. [^4]
| Preset | Meaning | Cron Expression |
|-----------|------------------------------------------------------|--------------------|
| None | Don’t schedule, use for exclusively “externally triggered” DAGs | |
| @once | Schedule once and only once | |
| @hourly | Run once an hour at the beginning of the hour | 0 * * * * |
| @daily | Run once a day at midnight | 0 0 * * * |
| @weekly | Run once a week at midnight on Sunday morning | 0 0 * * 0 |
| @monthly | Run once a month at midnight of the first day of the month | 0 0 1 * * |
| @yearly | Run once a year at midnight of January 1 | 0 0 1 1 * |
#### start_date
직관적이므로 생략.
#### catchup
start_date가 현재 날짜 기준으로 과거인 경우, start_date부터 현재 날짜까지의 워크플로우를 실행하는지 여부다. 예를 들어 start_date가 3주 전 일요일이고 schedule이 `@weekly`인 경우, DAG가 활성화되면 연속하여 3번 실행이 되는 식이다.
#### dagrun_timeout
타임아웃 설정으로 실행 시간이 일정 시간 초과하면 중지하고 실패로 처리하는 설정이다.
#### tags
DAG의 메타 정보다. 설정하면 검색하기 편하다.
#### params
DAG 실행할 때 입력 받는 변수로 key는 변수 이름이고 value는 디폴트 값이다.
## Operator 설정
```python
) as dag:
end = EmptyOperator(
task_id="end", # 태스크 이름
)
write_hello_world = BashOperator(
task_id="write_hello_world",
bash_command='echo "hello world!"'
)
write_hello_world >> end
this_will_skip = BashOperator(
task_id="this_will_skip",
bash_command='echo "this_will_skip"; exit 99;',
dag=dag,
)
this_will_skip >> end
```
[^1]: [Python Doc](https://docs.python.org/3/reference/simple_stmts.html#future)
[^2]: [Pendulum Introduction Page](https://pendulum.eustace.io/docs/#introduction)
[^3]: [Python Context Manager](https://wikidocs.net/231339)
[^4]: [Airflow 1.10.1 API Doc](https://airflow.apache.org/docs/apache-airflow/1.10.1/scheduler.html)