[Airflow] @task 데코레이터 사용하기

1. 파이썬 데커레이터

• Decorator는 장식하고 꾸미는 것을 의미하는데, 함수를 장식하는 것이라고 이해하시면 됩니다.
• 원래의 함수를 감싸서(Wrapping) 바깥에 추가 기능을 덧붙이는 방법입니다.

※ 함수를 감싼다는 것이 어떤 의미일까요?

• 파이썬은 함수 안에 함수를 선언하는 것이 가능하고
• 함수의 인자로 함수를 전달하는 것이 가능하며
• 함수 자체를 리턴하는 것이 가능합니다.

def outer_func(target_func):
	
    	# 내부 함수 정의
	def inner_func(): 
    		print("target 함수 실행 전입니다.")
        	target_func()
        	print("target 함수 실행 후 입니다.")
        
   	# 내부 함수 리턴
   	return inner_func

예시를 한 번 들어봅시다.

without decorator

파이썬 데코레이터를 사용하면?

python decorator

에어플로우에도 task decorator를 제공해줍니다.

 

2. Task 데커레이터

: 파이썬 함수 정의만으로 쉽게 task 생성을 할 수 있습니다.

task_decorator

decorator(@)를 사용하면 PythonOperator를 import 해올 필요도 없고, 실행을 시키기 위해서 py_task_1 >> py_task_2 이런식으로 연결된 구문을 작성해줄 필요도 없어집니다. 

단순히 py_task_1 과 같이 데코레이터에서 id값을 준 task만 써주면 끝입니다. 코드가 엄청 간결해질 수 있는 셈이죠.

에어플로우 공식 가이드에서도, PythonOperator를 import해서 DAG을 만드는 것보다, 그냥 TaskDecorator를 이용해서 만드는 것이 더 좋다고 나와있습니다.

나중에도 task간의 데이터를 공유하는 방법을 쓸 때도, TaskDecorator를 쓰면 한결 더 쉬워집니다.

 

그럼 바로 실습을 해보겠습니다. 먼저 새로운 dag파일을 아래와 같이 작성했습니다.

from airflow import DAG
from airflow.decorators import task
import pendulum

with DAG(
    dag_id = "dags_python_task_decorator",
    schedule="0 2 * * 1",
    start_date=pendulum.datetime(2023, 8, 1, tz="UTC"),
    catchup=False,
) as dag:
    
    @task(task_id="python_task_1")
    def print_context(some_input):
        print(some_input)

    python_task_1 = print_context("Run task_decorator!!")

에어플로우에서, 제대로 동작하는지 확인해보면 아래와 같습니다.

데코레이터를 씀으로써 PythonOperator 없이도, 특정 함수를 실행시킬 수 있다는 것을 볼 수 있습니다.

 

정리하자면,

 

파이썬 데코레이터를 쓰면, 

1) 함수를 직접 수정하지 않고도 wrapping 함수를 만듦으로써 함수의 변경을 최소화할 수 있습니다.

2) 함수를 호출하는 로직이 여러군데 있을 경우, 로직 부분을 일일히 찾아가서 모두 수정하지 않고도, 기존에 호출하던 로직 그대로 사용할 수 있습니다.

 

이와 마찬가지로, Airflow에서도 python decorator를 사용하는데요, 

이를 이용하면, Python Operator를 직접 import해서 만들지 않아도 task decorator를 쓰면 쉽게 만들 수 있습니다.  

이렇듯, task decorator 사용에 익숙해져야 합니다!