[Airflow] Python 함수 실행하기 | PythonOperator 사용

PythonOperator는 Apache Airflow에서 Python 함수를 실행하는 작업을 정의하는 데 사용되는 연산자이다. 이를 통해 Python 함수를 호출하여 데이터 처리, 계산, 또는 사용자 지정 작업을 수행할 수 있다. 아래는 PythonOperator를 사용하는 간단한 예제이다.

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

# DAG 정의
dag = DAG(
    'python_operator_example',
    description='Example DAG with PythonOperator',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 1, 1),
    catchup=False,
)

# Python 함수 정의
def my_python_function():
    print("Hello, Airflow!")
    print("Current date:", datetime.now())

# PythonOperator 정의
python_task = PythonOperator(
    task_id='python_task',
    python_callable=my_python_function,
    dag=dag,
)

# DAG 간의 의존성 설정
# 다른 작업들과의 의존성이 있다면 여기에 추가

# DAG 실행
if __name__ == "__main__":
    dag.cli()

• 이 예제에서 PythonOperator를 생성할 때 task_id로 작업의 고유한 식별자를 설정하고, python_callable 매개변수에는 실행할 Python 함수를 전달
• 이 함수는 Airflow 실행 환경에서 실행되며, 함수의 출력 및 오류는 자동으로 Airflow 로그에 기록됨
• DAG 내에서 작업 간의 의존성을 설정하려면 필요한 작업의 task_id를 사용하여 의존성 연산자를 정의하면 됨
• 이는 set_upstream, set_downstream, 또는 >>, << 연산자를 사용하여 수행할 수 있음