Airflow 기본

Airflow Concepts-Architecture

Architecture Overview
- workflow를 실행할 수 있도록 만드는 플랫폼으로 DAG를 기반으로 한다.
- DAG에 의존성을 등록된 각각의 work를 task라고 한다.
- Scheduler: triggering workers & submitting tasks
- Executor: task의 실행을 담당 → task를 workers에 등록(push)
- Webserver: UI를 제공하는 서버
- DAG files: scheduler와 executor가 읽는 파일
- Metadata database: state를 저장하기 위해서 workers, scheduler, webserver가 접근
Workloads
- Operator: 사전에 정의된 task
- Sensors: Operators의 하위 클래스로 외부 이벤트가 발생할 때까지 기다린다
  - JS의 이벤트 리스너
- TaskFlow-decorator @task: 파이썬에서 패키지로 지원하는 데코레이터
Control flow
- 디펜던시 설정 관계
- upstrea/downstream

Airflow Concepts-Worklaods

DAGs
- Task들이 의존성을 기반으로 조직된 Airflow의 핵심 개념
- 어떤 task가 의존성을 가지는지와 얼마나 자주 DAG가 실행되는 지를 정의할 수 있다.
- Declaring a DAG
  - with 구문을 이용하는 방법
```
with DAG("dag_name") as dag:
    t = DummyOperator(task_id="task")
```
  - 생성자를 이용하는 방법
```
my_dag = DAG("dag_name")
op = DummyOperator(task_id="task", dag=my_dag)
```
- Task dependencies
  - chan(*args)를 사용
  - args[0] >> args[1] >> ... >> args[n-1]
- Loading DAGs
  - config파일에 정의된 DAG_FOLDER를 기반으로 로딩
  - 가장 top level에서 로드된 것만 사용할 수 있다.
  - global 정의 → 함수 내 정의(로드❌)
- Running DAGs
  - API를 통해 실행
  - DAG 내부에서 정의
  - schedule_interval: 시간을 베이스로 실행이된다.
Tasks
- Airflow에서 실행되는 기본 유닛
- DAG를 통해 정렬이 정렬이 된다.
- none → scheduled → queued → running → success
- Timeouts
  - 런타임의 최대 허용시간을 정할 수 있다.
  - exception을 불온다.
- SLAs
  - Exception대신 email 메시지를 전송하는 timeout의 또다른 형태
Tasks-Operators
- 정의된 task를 의미한다.
- Jinja templating을 이용해 만들 수 있다.
Tasks-Sensors
- 기다리다가 실행이 되도록하는 task를 만들기 위함
Scheduler
- 모든 task들과 DAG들을 관리한다.
- Dependecy가 완료가 되면 task instance를 촉발 시킨다.
- 작동 원리
  - Subprocess 아래에서 계속 대기(spin)
  - 특정 DAG 디렉토리에 있는 DAG와 동기활를 한다.
- airflow scheduler 명령어를 통해 제어를 할 수 있다.
  - 커스터마이징은 airflow.cfg에서 진행한다.
- Executor를 통해 준비된 task를 실행한다.
Template
- 파일을 실행하기 위한 코드를 만드는 규칙?
```
{% for i in range(5) %}
    echo "{{ ds }}"
    echo "{{ macros.ds_add(ds, 7)}}"
    echo "{{ params.my_param }}"
{% endfor %}
```
- 위의 파일은 bash 스크립트를 만들기 위해 Jinja를 활용한 template이다. {{ ds }}의 경우는 macro로 date time을 의미한다.

Tutorial(DAG definitaion file)

Module import

필요한 모듈 가져오기

from datetime import timedelta
from textwrap import dedent

# The DAG object; we'll need this to instantiate a DAG
from airflow import DAG

# Operators; we need this to operate!
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago

Default Arguments

각각의 task constructor에게 어떤 인자를 넣어줄 것인지를 정의 할 수 있다.

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['[email protected]'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
    # 'wait_for_downstream': False,
    # 'dag': dag,
    # 'sla': timedelta(hours=2),
    # 'execution_timeout': timedelta(seconds=300),
    # 'on_failure_callback': some_function,
    # 'on_success_callback': some_other_function,
    # 'on_retry_callback': another_function,
    # 'sla_miss_callback': yet_another_function,
    # 'trigger_rule': 'all_success'
}

overriding도 가능하다.
airflow.models.BaseOperator을 참고하자

airflow.models - Airflow Documentation

Instantiate a DAG

기본인자와 마찬가지로 각각의 task에 포함될 DAG 객체를 생성하는 부분이다.

with DAG(
    'tutorial', # setting dag_id
    default_args=default_args,
    description='A simple tutorial DAG',
    schedule_interval=timedelta(days=1),
    start_date=days_ago(2),
    tags=['example'],
) as dag:

dag_id로 문자열을 그대로 넘겨준다.

Tasks
```
t1 = BashOperator(
    task_id='print_date',
    bash_command='date',
)
```
- Operator의 인스턴스를 task라고 부르며 task_id가 구분 인자로 필수적으로 필요하다.
- Task는 반드시 task_id 혹은 owner가 반드시 필요하다.

Templating with Jinja

templated_command = dedent(
    """
{% for i in range(5) %}
    echo "{{ ds }}"
    echo "{{ macros.ds_add(ds, 7)}}"
    echo "{{ params.my_param }}"
{% endfor %}
"""
)

code logic: {% %} block
referenced parameters: {{ }} block

t3 = BashOperator(
    task_id='templated',
    depends_on_past=False,
    bash_command=templated_command,
    params={'my_param': 'Parameter I passed in'},
)

bash_command에 필요 및 실행할 bash 스크립트를 넣어주며 → templated_command.sh
params를 이용해서 dict 객체를 전달한다.
유저가 직접 macro를 생성해서 사용할 수 있다.

Adding DAG and Tssks documentation
Setting up Dependencies