[Airflow] DAGs 사용 방법 정리

DAGs

airflow에서는 워크플로우를 DAG(Directed Acyclic Graph)로 관리. 

DAGs : 비순환, 방향성을 가지고 있는 그래프를 뜻함.

DAGs

Airflow 웹서버에서 DAGs Task 확인. 

DAGs - Operator의 모음

DAGs는 Operator(Task) 의 모음이다. 

개별 Task 상태 값을 확인하여 재실행, Failed 마킹, Success 마킹 등의 여러가지 개별 작업을 수행할 수 있다. 

 

DAGs 작성방법 (주의점)

Scope

파이썬 파일 내에 DAG는 전역 스코프에 존재해야 함.

dag_1 = DAG('this_dag_will_be_discovered')

def my_function():
  dag_2 = DAG('this_dag_will_not')
  
my_function()

위와 같이 작성하게 되면 dag_1만 airflow에 로드. dag2는 로드되지 않음. 

Default arguments

파라미터로 default_arg를 입력하여 operator에 적용. 

default_args = {
  'start_date' : datetime(2016, 1, 1),
  'owner' : 'airflow'
}

dag = DAG(
	dag_id='example_dag_tag',
	default_args=default_args,
	schedule_interval='0 0 * * *',
	tags=['example'] # airflow 1.10.8, 1.10.9 버전부터 tag 기능 생김. 
)
op = DummyOperator(task_id='dummy', dag=dag)
print(op.owner) # Airflow

Operators

DAG는 Operator의 모음(전후 관계 설정) 

즉, airflow의 작업은 [ 1. DAG 객체 생성 -> 2. Operator 활용한 Task 작성 -> 3. Task를 연결 ] 이뤄져 있음. 

Airflows의 Operator 종류

내부 :

• BashOperator : bash 명령 실행
• PythonOperator : 임의의 python function 호출
• EmailOperator : Email 전송

외부 : 

• SimpleHttpOperator : HTTP request 전송
• MySqlOperator, SqliteOperator, PostgresOperator, MsSqlOperator, OracleOperator, JdbcOperator, etc. : SQL 명령을 실행
• Sensor : 특정 시간, 파일 db row를 Poll (wait)하는 오퍼레이터