[4주차] Airflow Deepdive - Airflow로 데이터 파이프라인 만들기 (1)

Hello World 예제 프로그램, Hello World 살펴보기 프로그램을 설명하고 Web UI와 터미널에서 실행해 보자

● 2개의 태스크로 구성된 데이터 파이프라인 (DAG)

 ○ print_hello: PythonOperator로 구성되어 있으며 먼저 실행

 ○ print_goodbye: PythonOperator로 구성되어 있으며 두번째로 실행

dag = DAG(
 dag_id = "helloWorld",
 start_date = datetime(2022,6,15),
 catchup=False,
 tags=['example'],
 schedule = '0 2 * * *',
 default_args=default_args
)

 

Operators - PythonOperator

● Airflow Python Operator는 Airflow에서 사용되는 작업(task)을 정의하는 Python 코드를 실행하고 관리하는 데 사용되는 요소

● 각 Python Operator는 특정 작업 또는 작업 단계를 정의하는 Python 함수나 메서드를 참조

● 이러한 Operator를 사용하면 Airflow DAG (Directed Acyclic Graph)에서 작업 단계를 정의하고 관리할 수 있다.

● Operator를 표현하는 방식은 2가지가 있음.

● Task Decorator를 사용하면 훨씬 더 프로그램이 직관적

Task 파라미터들

● 여기에 지정되는 인자들은 모든 태스크들에 공통으로 적용되는 설정이 됨

● 뒤에서 DAG 객체를 만들 때 지정함

 

중요한 DAG 파라미터 (not task 파라미터)

with DAG(
 dag_id = 'HelloWorld_v2',
 start_date = datetime(2022,5,5),
 catchup=False,
 tags=['example'],
 schedule = '0 2 * * *',
 default_args=default_args
) as dag:

● max_active_runs: # of DAGs instance

● max_active_tasks: # of tasks that can run in parallel

● catchup: whether to backfill past runs

● DAG 파라미터 vs. Task 파라미터 차이점 이해가 중요

● 위의 파라미터들은 모두 DAG 파라미터로 DAG 객체를 만들 때 지정해주어야함

 

How to Trigger a DAG - From the Airflow Web UI

● Airflow Web UI에서 ▶ 클릭

 

How to Trigger a DAG - 터미널에서 실행 (1)

● Airflow 컨테이너 중 하나로 로그인:

 ○ docker ps # Worker의 컨테이너ID를 찾을 것(ex: 9b7878041c3b)

 ○ docker exec -it 컨테이너ID sh

 

● 로그인 후 다음 명령 실행

 ○ airflow dags list

 ○ airflow dags test DAG이름 날짜

 ○ airflow tasks list DAG이름

 ○ airflow tasks test DAG이름 Task이름 날짜 # test vs. run

  ■ 날짜는 YYYY-MM-DD

 

● start_date보다 과거인 경우는 실행이 되지만 오늘 날짜보다 미래인 경우 실행 안됨

● 이게 바로 execution_date의 값이 됨

 

How to Trigger a DAG - 터미널에서 실행 (2)

docker ps
docker exec -it 컨테이너ID sh
airflow dags list
airflow tasks list HelloWorld
airflow tasks test HelloWorld print_hello 2020-08-09
airflow dags test HelloWorld 2019-12-08
airflow dags backfill MySQL_to_Redshift_v3 -s 2019-01-01 -e 2019-12-31