Airflow - bash dag샘플

from airflow.models.dag import DAG

import datetime

import pendulum

from airflow.operators.bash import BashOperator

with DAG(

    dag_id="dags_bash_operator", # web ui에 보여지는 이름, py파일명과는 무관.

    schedule="0 0 * * *",

    #start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),

    start_date=pendulum.datetime(2021, 1, 1, tz="Asia/Seoul"),

    catchup=False,              # True 일 경우 start_date인 2021.1.1 ~ 현재 까지 전부 실행됨. 기본적으로 False

    dagrun_timeout=datetime.timedelta(minutes=60), # 60분 이상 실행 시 실패로 처리

    tags=["ex", "ex2"], # web ui 태그

    params={"example_key": "example_value"},

) as dag:

      bash_t1 = BashOperator(

        task_id="bash_t1",

        bash_command="echo test1",

    )

     

      bash_t2 = BashOperator(

        task_id="bash_t2",

        bash_command="echo $HOSTNAME",

    )  

     

      bash_t1 >> bash_t2 # 실행 순서

 

bash_t2의 로그를 보면 $HOSTNAME 결과로 airflow의  worker container id가 뜨게 된다.