[Airflow] - 3. Dags(Python operator)

개발/Airflow

[Airflow] - 3. Dags(Python operator)

Dortmoot 2022. 7. 8. 17:36

Dag 파일을 만들어보고 Workflow를 Pythonoperator로 구현하는 예시를 보여드릴 예정입니다.

틀린 내용이 있을 수 있으니 상세한 부분 Document를 참고하시는 것을 추천 드립니다.

1. Dag 생성

Dag file을 생성하기 위해서는 일단 Ubuntu의 Dags 폴더 File을 들어갑니다.

왜 Dag 폴더를 이용하는 가는 실제 Docker Container가 실행되면 dags / logs / plugins 폴더를 Mount 하여 가져갑니다.

( 이부분은 Docker Container를 직접 접속하셔서 /opt/airflow/dags 에 들어가시면 확인 할 수 있습니다. )

후에 Dag를 만들 Test.py 파일을 만듭니다.

$ cd /dags
$ sudo vi test.py

이제부터는 어떤 방식으로 작업해야 할지에 대해 설명 드리겠습니다.

How Dag Create?

1. Define Default_args

owner(만든 사람), start_date(언제부터 시작하게 할것인지) 값을 설정하여 줍니다.

2. Create Dag Object

Dag_id, Schdule_interval 값을 설정하여 줍니다.

3. Create Task

Operator를 이용하여 Task 생성하여 줍니다.

4. Connect Task

'<<' , '>>' 를 사용하여 Task를 연결하여 줍니다.

2. Dag Code

Parameter가 무슨 의미를 하는지 살펴보겠습니다.

airflow module을 통하여 Dag 와 Pythonoperator를 부를 예정입니다.

실제 Scheduler에게 어떤 시간에 동작하게 할 것인지 필요함으로 datetime module도 부를 예정입니다.

추가적으로 Webserver에서 Log를 찍으면서 올바른 결과가 사용되지는도 볼 예정입니다.

코드를 먼저 보는 편이 이해하기에 쉬우리라 생각되어 코드를 먼저 보여드리겠습니다.

from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import logging

#dag option init
dag = DAG(
        dag_id='uk_dag_test_1', #id option
        schedule_interval='*/60 * * * *', #(매 10분마다 실행) / schedule_interval='@daily' (time 주기)
        start_date = datetime(2022,7,8), #time start / 금일부터 실행하고 싶다면 금일 전날짜로 set
        tags=['uksung_test']
        )

#Python
def test(**context):
    print('test function start')
    parameter=context["params"]["parameter1"]
    logging.info(parameter)
    return parameter

def test_parameter_trnasform(**context):
    transform_data = context['task_instance'].xcom_pull(task_ids='kimuksung_test')
    logging.info(transform_data)
    return True

#DAG Task
test_job = PythonOperator(
    #task id setting
	task_id = 'kimuksung_test',
	#python_callable param points to the function you want to run 
	python_callable = test,
	#dag param points to the DAG that this task is a part of
	dag = dag,  
    #parameter setting
    params = {'parameter1':'pubg'},
    #parameter on/off
    provide_context=True
    )

test_transform = PythonOperator(
    #task id setting
	task_id = 'kimuksung_transform',
	#python_callable param points to the function you want to run 
	python_callable = test_parameter_trnasform,
	#dag param points to the DAG that this task is a part of
	dag = dag,
    #parameter on/off
    provide_context=True
    )

test_job >> test_transform

이제 Dag parameter option을 살펴보겠습니다.

1. dag_id = Webserver에서 보여지는 Dag_id

2. tags = Webserver에서 보여지는 Tag

3. start_time = Scheduler가 해당 Dag를 언제부터 실행시킬 것인지

4. scheduler_interval = scheduler가 얼마만큼의 Time 주기마다 Dag를 실행할 것인지

다음으로 PythonOperator를 살펴보겠습니다.

1. task_id = Dag 내에서 실행될 Task_id 입니다.

2. python_callble = Task가 실행될 때 호출하는 Function

3. dag = 위에서 설정한 Dag 설정값을 호출

4. provide_context = Parameter 전달 여부 ( True / False )

5. params = Parameter value

제일 아래에 있는 test_job >> test_transform 는 Task가 실행될 순서를 보여줍니다.