[Airflow] op_args와 op_kwargs 사용하기

Airflow는 데이터 파이프라인 작업을 DAG(Directed Acyclic Graph)로 정의하여 작업 순서를 지정할 수 있습니다. 이때 PythonOperator를 통해 Python 함수를 task로 사용할 수 있는데, 이 함수에 인자를 전달할 때 op_args와 op_kwargs 파라미터를 사용합니다.

---

op_args: 위치 인자 전달

op_args는 위치 기반 인자(positional arguments)를 전달하기 위한 리스트 형태의 파라미터입니다. 이를 통해 함수에 순서대로 값을 전달할 수 있습니다.

 

예시

def my_task_with_args(arg1, *args):
    print(f"arg1: {arg1}")
    print(f"Additional args: {args}")
    
PythonOperator(
    task_id='my_task_with_args',
    python_callable=my_task_with_args,
    op_args=['hello', 'extra', 'args'],
    dag=dag,
)

설명: arg1에는 hello가, *args에는 ('extra', 'args')가 할당됩니다.

---

op_kwargs: 키워드 인자 전달

op_kwargs는 딕셔너리 형태로 키워드 인자(keyword arguments)를 전달할 수 있습니다. 함수에 키와 값을 매칭해 전달할 수 있는 방식입니다.

def my_task(arg1, arg2):
    print(f"arg1: {arg1}, arg2: {arg2}")

PythonOperator(
    task_id='my_task',
    python_callable=my_task,
    op_kwargs={'arg1': 'hello', 'arg2': 'world'},
    dag=dag,
)

설명: op_kwargs에서 {'arg1': 'hello', 'arg2': 'world'}를 전달해 각 변수에 값을 할당합니다