Airflow | Python Operator에서 Template 변수 사용

Airflow 마스터 클래스 강의 수강, 개인 공부 후 기록합니다.

공식 문서 확인

Python 오퍼레이터에는 어떤 파라미터에 template 변수를 쓸 수 있는지 알아보고 사용해본다. 먼저 airflow 공식 문서를 확인한다.

 

https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/operators/python/index.html

 

Bash Operator에서 Template 변수 사용에서와 마찬가지로 Parameters, template_fields를 모두 확인한다.

templates_dict 파라미터에만 (templated)가 있고 이외에는 없다.

template_fields에는 templates_dict 이외에도 op_args, op_kwargs가 추가로 쓰여져 있는 것을 확인할 수 있다.

 

Python operator에서 templates_dict, op_args, op_kwargs 파라미터에 template 변수를 사용할 수 있다.

Template 변수 사용

Python operator에서 template 변수를 2가지 방법을 통해 사용해본다.

1. template 변수가 사용 가능한 파라미터에 operator를 선언할 때 template 변수를 직접 넣기
2. kwargs에 template 변수들을 자동으로 제공해주는 점 이용(airflow 날짜 개념)

 

1. op_kwargs에 template 변수를 직접 넣기

def python_function1(start_date, end_date, **kwargs):
    print(start_date)
    print(end_date)


python_t1 = PythonOperator(
    task_id="python_t1",
    python_callable=python_function1,
    op_kwargs={"start_date":"{{ data_interval_start | ds }}", "end_date":"{{ data_interval_end | ds }}"}
)

key, value 형식을 사용해 op_kwargs를 설정할 때 value에 template 변수를 사용한다.

 

2. kwargs에 이미 정의되어 있으므로 꺼내서 사용

@task(task_id="python_t2")
def python_function2(**kwargs):
    print(kwargs)
    print("ds:" + kwargs["ds"])
    print("ts:" + kwargs["ts"])
    print("data_interval_start:" + str(kwargs["data_interval_start"]))
    print("data_interval_end:" + str(kwargs["data_interval_end"]))
    print("task_instance:" + str(kwargs["ti"]))

python operator는 kwargs에 template 변수들을 자동으로 제공한다. 따라서 kwargs에 template 변수를 key 값으로 입력해 그에 해당하는 값을 가져올 수 있다.