[Airflow] airflow ui에서 arguments 전달하기

개요

Airflow의 Task들은 스케줄링으로 관리하고 있다.

하지만 단발성으로 실행되어야하는 Task들도 존재한다.

초기 데이터 적재 혹은 재적재 이슈로 인해..

이러한 경우에 Arguments을 전달하고 싶을 때 어떻게 해야할까 찾아보았다.

진행

BashOperator, PythonOperator 두가지 방법으로 테스트해보았다.

from datetime import datetime, timedelta
from airflow import DAG
from pendulum.tz.timezone import Timezone
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator
kst = Timezone('Asia/Seoul')
default_args={
    'owner' : 'test',
    'retries' : 1,
    'retry_delay' : timedelta(minutes=1)
}
def print_arguments(**kwargs):
    message = kwargs['dag_run'].conf
    print(message)
with DAG(
    dag_id='Argument_Test',
    description='Dag to test how to send arguments',
    default_args=default_args,
    start_date=datetime(2024, 7, 11, tzinfo=kst),
    schedule_interval=None,
    tags=['TEST']
) as dag:
    ## bash operator
    bash_task = BashOperator(
        task_id="bash_task",
        bash_command="echo '{{ dag_run.conf }}'",
        dag=dag
    )
    ## python operator
    python_task = PythonOperator(
        task_id="python_task",
        python_callable=print_arguments,
        provide_context=True, ## 꼭 추가하기
        dag=dag
    )
    bash_task >> python_task

 

위처럼 스크립트 작성해서 Argument_Test라는 DAG를 생성했다.

 

실행버튼 클릭해서 config와 같이 trigger 한다는 Trigger DAG w/ config를 선택한다.

 

 

JSON 형식으로 message를 입력한다.

 

BashOperator로 실행된 결과

 

PythonOperator로 실행된 결과

내가 입력한 message를 잘 받아서 print 찍은 것을 확인할 수 있다.

 

이제 단발성으로 실행되어야하는 Task 발생 시, airflow를 활용할 수 있겠다. 굳굳

 

지금 나는 크롤링을 하고 있는데, 단발성으로 실행해야하는 경우

1. 특정 기간의 데이터를 크롤링 하고 싶은 경우

2. 특정 도메인의 데이터를 크롤링 하고 싶은 경우

그럼 특정 기간, 특정 도메인은 arguments로 넘겨서 진행하면 되니까 훤씬 가볍게 DAG를 trigger할 수 있다.