[Week10 Airflow] TIL 43일차 Airflow DAG, task의 정의와 사용

Ⅰ. 용어 설멍

1. DAG란?

Directed Acyclic Graph(방향이 있는 비순환 그래프)로 Airflow의 작업의 단위를 말한다.

DAG는 여러 task로 구성되어 있는데, task의 실행방법, 실행순서, 재시도 횟수, 시간 초과 등의 정보를 담고 있다.

2. Task란?

Workflow의 최소 실행 단위이다. 원칙적으로는 하나의 Operator을 가진다(단, 특별한 경우에 여러 Operator를 사용하고 싶다면, PythonOperator나 BranchPythonOperator를 사용하여 Python 함수를 정의하고, 해당 함수 내에서 여러 작업을 수행하도록 구현해야 한다.)

3. Operator란?

Operator는 Airflow에서 어떤 작업을 수행할지를 정의하는 클래스이다. Task의 템플릿이라고 생각하면 된다.

오퍼레이터의 종류는 다음과 같다.

Operators — Airflow Documentation

 

종류

BashOperator:

명령 줄 명령어나 스크립트를 실행

PythonOperator: + @test와 함께 사용!

Python 함수를 실행하는 Operator로, 사용자 정의 Python 코드를 실행

DummyOperator:

아무 작업도 수행하지 않는 Operator로, 주로 DAG 내에서 조건부 논리를 구성할 때 사용

HttpSensor:

HTTP 엔드포인트의 상태를 확인하여 센서 역할을 하는 Operator

BranchPythonOperator:

Python 함수의 반환 값에 따라 다른 경로로 DAG를 분기

SubDagOperator:

SubDAG를 호출하는 Operator로, 하위 DAG를 실행할 때 사용

TriggerDagRunOperator:

다른 DAG를 실행하기 위해 사용되며, 외부 DAG를 트리거할 수 있음

S3FileTransformOperator:

AWS S3 파일을 변환하는 Operator로, 주로 데이터 처리 작업에 사용

EmailOperator:

이메일을 보내는 Operator로, 작업 완료 또는 오류 알림에 사용

SqlSensor:

SQL 질의 결과를 확인하여 센서 역할을 하는 Operator

DockerOperator:

Docker 컨테이너 내에서 명령을 실행하는 Operator

KubernetesPodOperator:

Kubernetes 클러스터 내에서 파드를 실행하는 Operator로, 컨테이너 오케스트레이션에 사용

 

 

Ⅱ.  DAG를 선언하는 3가지 방법

0. DAG 파라미터 종류

필수 매개변수

1) dag_id

DAG의 이름으로, 고유하게 식별 가능한 문자열이어야 한다.

2) start_date

DAG의 시작 날짜 및 시간을 나타낸다.

3) schedule

얼마나 자주 실행될지를 결정하는 DAG 실행 주기로 timedelta 객체로 전달해야 한다.

형식

" Minute[0-59] Hour[0-23] Day[1-31] Month[1-12] Week[1-7 start from monday] "

혹은

@none, @once, @hourly, @daily, @weekly, @monthly, @yearly

선택적 매개변수

1) default_args

DAG와 연관된 공통 적용 설정에 관한 정보이다. 딕셔너리 형태로 전달해야 한다.

2) catchup

startdate가 과거이고, DAG활성화를 그 이후에 했을 때, 이전 실행들을 잡아내기(catch up) 여부를 결정하는 매개변수이다.

이전의 밀린 날짜들마다 schedule대로 DAG를 실행할 것이다. 기본 설정은 True이다.

Full refresh는 의미가 없고 Incremental update를 수행하는 DAG에만 의미가 있다. 

3) tags

태그에 대한 정보로 해당 DAG의 태그 목록을 전달한다.

이후에 많은 DAG가 생성되면, 보통 태그로 묶어서 관리한다(검색)

4) end_date

DAG의 종료 날짜에 대한 매개변수이다.

5) max_active_runs

동시에 실행되는 최대 DAG 인스턴스 수를 제한합니다. (backfill을 수행하는 경우 여러 개의 DAG를 실행해야 한다. 이때, 병렬로 여러 DAG를 수행한다.) 

6) max_active_tasks

동시에 실행되는 최대 task 수를 제한합니다. (일렬로 task가 배치된 경우 의미가 없으나 병렬 task가 존재하는 경우 의미가 있다.)

5) 번, 6) 번의 경우 아무리 많은 수의 DAG, task를 지정하여도, 실제로는 airflow worker 노드에 적용된 CPU개수 이상으로 할당되진 않는다. 

7) retries

실패 시 다시 시도하는 횟수

8) retry_delay

실패 시 다시 시도할 때의 딜레이에 대한 시간(분 단위)

9) on_failure_callback 

task 실패 시 지정된 함수가 호출

보통, slack과 연동하여 에러를 명시적으로 알 수 있게 한다.

10) on_success_callback

성공 시 지정된 함수가 호출

 

1. Context Manager 사용

: with 문을 사용하여 코드 블록을 감싸는 파이썬 객체

 import datetime

 from airflow import DAG
 from airflow.operators.empty import EmptyOperator

 with DAG(
    dag_id='DAG의 이름'
    start_date=datetime(년도, 월, 일)
    schedule="@daily",
 ):
 오퍼레이터이름(task_id="TASK의 ID")

 

2. Standard constructor 사용

: 오퍼레이터에 dag매개변수를 이용하는 것

 import datetime

 from airflow import DAG
 from airflow.operators.empty import EmptyOperator

 my_dag=with DAG(
    dag_id='DAG의 이름'
    start_date=datetime(년도, 월, 일)
    schedule="@daily",
 )
 오퍼레이터이름(task_id="TASK의 ID", dag=my_dag)

 

3. @dag 데코레이터 사용

import datetime

from airflow.decorators import dag
from airflow.operators.empty import EmptyOperator


@dag(start_date=datetime.datetime(년도, 월, 일), schedule="@daily")
def generate_dag():
    오퍼레이터(task_id="TASK의 ID")

generate_dag()

 

Ⅲ. 실습

실습은 서버에 airflow를 직접 설치하는 경우, docker을 이용하는 경우 두 가지로 진행하였다.

앞의 경우는 AWS EC2 서버를 이용하였고 뒤의 경우는 GCP VM 엔진에 docker를 설치하여 사용한다.

AWS, GCP 모두 경험하고 싶었기 때문이다.

 

0. dag 기본 저장 위치 확인 airflow.cfg

aws의 경우

dags_folder = /var/lib/airflow/dags

 

docker의 경우

dags_folder = /docker/airflow-setup/dags

 

해당 위치에 dag 들을 위치해야 Airflow가 자동으로 dag로 인식을 하게 된다.

 

1. AWS EC2 계정에 직접 airflow를 설치하는 경우

1) EC2 계정 로그인

ssh -i "airflow-dev.pem" ubuntu@~

2) DAG 목록 확인

airflow dags list

3) DAG의 task목록 확인

airflow tasks list DAG이름

4) task 실행

airflow task [text | run] DAG이름 task이름 날짜(YYYY-MM-DD)

• test : task의 실행이 메타데이터 DB에 기록 O
• run : task의 실행이 메타데이터 DB에 기록 O

 

2. Docker 실습 시(GCP 이용)

1) Airflow schedular에 로그인

- airflow schedular 컨테이너 ID 확인(airflow-setup-airflow-schedular-1)

docker ps

- airflow schedular 컨테이너에 쉘스크립트 띄우기

docker exec -it 컨테이너 ID sh

(이후의 과정은 EC2와 동일하다)

2) DAG 목록 확인

airflow dags list

3) DAG의 task목록 확인

airflow tasks list DAG이름

4) task 실행

airflow task [text | run] DAG이름 task이름 날짜(YYYY-MM-DD)

• test : task의 실행이 메타데이터 DB에 기록 O
• run : task의 실행이 메타데이터 DB에 기록 O