[Airflow] Task 간 의존성 설정

Task 간 의존성

여러 개의 Task가 존재할 때 의존성을 결정하는 방법을 정리한다.

 

 

set_downstream() / set_upstream()

Task 간 의존성은 set_downstream 또는 set_upstream과 같은 함수로 나타낼 수 있다.

• set_downstream : Task 실행 후에 수행할 task를 설정한다.
• set_upstream : Task 실행 전에 수행할 task를 설정한다.

다음과 같이 두 개의 Task를 가진 DAG가 존재한다고 하자.

# DAG 인스턴스화
dag = DAG(
    'helloworld',
    default_args=default_args,
    description='print "hello, world!"',
    schedule_interval=timedelta(days=1),
)

def hello_world():
    print("world!")

# task
t1 = BashOperator(
    task_id='echo_hello',
    bash_command='echo "Hello"',
    dag=dag,
    # task_concurrency=2
)

t2 = PythonOperator(
    task_id='print_world',
    python_callable=hello_world,
    dag=dag
)

이 DAG의 Tree는 아래와 같다. 실행 시 두 Task가 동시에 실행한다. 

이것을 echo_hello 실행 후에 print_world 실행하도록 변경하고 싶다면 아래와 같이 코드를 추가한다.

t1.set_downstream(t2)

# 또는
t2.set_upstream(t1)

수정 후 DAG Tree를 확인하면 아래와 같이 변경된 것을 확인할 수 있다.

 

 

Shift 연산자

Task 간 의존성은 >>, <<과 같은 Shift로 연산자로 표현할 수 있다.  >>가 set_downstream, <<가 set_upstream에 해당한다.

이번에는 print_world 실행 이후에 echo_hello를 실행하고자 한다. 이 경우에는 아래와 같이 작성할 수 있다.

t2 >> t1

# 또는
t1 << t2

순서가 변경된 모습을 확인할 수 있다.

 

 

 

참고 문서

https://velog.io/@jjongbumeee/Airflow5