Airflow

Airflow란 에어비앤비에서 개발한 워크플로우 스케줄링, 모니터링 플랫폼 입니다.

현재 아파치의 탑레벨 프로젝트에 등록 되어 있으며

Hadoop echosystem 에서 특히 python기반으로 개발 되었기 때문에 Pyspark와 함께 많이 사용되는 툴 입니다.

테스트 서버에 구축 이 완료 되었으며 아래와 같은 시나리오로 개발/테스트 하였습니다.

총 3개의 Pyspark job을 사용하여 워크플로우를 구성 하였습니다.

먼저 csv 형식의 파일을 Postgresql DB에  A테이블에 적재 후 해당 작업 종료 후  2번 작업이 실행 되며 적재된 A테이블의 데이터를 Maria DB 의 A테이블로 ETL 한 후 마지막으로 3번째 작업인 Maria DB 의 A테이블을 Hdfs에 orc 확장자로 Append 하며 해당 워크플로우는 종료 됩니다.

 

Airflow 워크플로우 소스

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.dates import days_ago
 
default_args = {
    'owner': 'spidyweb',
    'retries': 0,
    'retry_delay': timedelta(seconds=20),
    'depends_on_past': False
}
 
dag_Spark = DAG(
    'dag_Spark_id',
    start_date=days_ago(2),
    default_args=default_args,
    schedule_interval='@once',
    catchup=False,
    is_paused_upon_creation=False,
)
 
cmd0='spark-submit /root/postcsv.py'
cmd1='spark-submit /root/post_to_maria.py'
cmd2='spark-submit /root/maria_hadoop.py'
 
task_start = DummyOperator(
    task_id='start',
    dag=dag_Spark,
)
 
task_next = BashOperator(
    task_id='_post',
    trigger_rule='all_success',
    dag=dag_Spark,
    bash_command=cmd0,
)
 
task_next1 = BashOperator(
    task_id='_maria',
    trigger_rule='all_success',
    dag=dag_Spark,
    bash_command=cmd1,
)
 
task_next2 = BashOperator(
    task_id='_hadoop',
    trigger_rule='all_success',
    dag=dag_Spark,
    bash_command=cmd2,
)
 
task_finish = DummyOperator(
    task_id='finish',
    trigger_rule='all_done',
    dag=dag_Spark,
)
 
task_start >> task_next >> task_next1 >> task_next2 >> task_finish

구동 화면