[Apache Airflow] Cron Schedule

Cron Schedule

 

Cron Schedule은 다음과 같이 task가 실행되어야 하는 시간 (주기)을 정하기 위한 다섯개의 필드로 구성된 문자열이다.

 

{분} {시} {일} {월} {요일}

 

Airflow의 모든 DAG는 실행 주기를 결정해야 하며, 일반적으로 Cron Schedule을 활용하여 DAG이 실행되어야 할 주기를 명시한다.

 

특수 문자

 

특수 문자	설명
*	모든 값
-	범위 지정
,	여러 값 지정
/	증가값 지정
L	마지막 값 (일, 요일에만 설정 가능) ※ 일에 L입력시 해당 월의 마지막 일 의미 ※ 요일에 L 입력시 토요일 의미
#	몇 번째 요일인지 지정

 

Cron Schedule	설명	비고
15 2 * * *	매일 02시 15분
0 * * * *	매시 정각
0 0 1 * *	매월 1일 0시 0분
10 1 * * 1	매주 월요일 1시 10분	0: 일요일, 1: 월요일, 2: 화요일, 3: 수요일, 4: 목요일, 5: 금요일, 6: 토요일
0 9-18 * * *	매일 9시부 터18시까지 정각 마다
0 1 1,2,3 * *	매월 1일, 2일 3일만 1시에
*/30 * * *	삼십분마다 (0분, 30분)
10-59/30 * * * *	10분부터 삼십분마다 (10분, 40분)
10 1 * * 1-5	평일만 01시 10분
0 */2 * * *	2시간마다 (0시, 02시, 04시 …)	1-23/2: 1시부터 2시간마다
0 0 */2 * *	짝수일 0시 0분
10 1 L * *	매월 마지막 일 01시 10분
10 1 * * 6#3	매월 세번째 토요일 01시 10분
None	주기 없음 (수작업으로만 실행되는 DAG)

 

Task 연결

 

DAG는 1개 이상의 Task를 가지며 일반적으로 Task는 선후행 관계를 가진다. 선후행 관계를 표시하는 방법에는 여러 방법이 존재하지만 가장 일반적인 방법은 >>와 << 기호를 이용한 방법이 있다.

 

from airflow import DAG
import datetime
import pendulum
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="dags_bash_operator",
    schedule="0 0 * * *",
    start_date=pendulum.datetime(2024, 6, 1, tz="Asia/Seoul"),
    catchup=False
) as dag:
    bash_t1 = BashOperator(
        task_id="bash_t1",
        bash_command="echo whoami",
    )

    bash_t2 = BashOperator(
        task_id="bash_t2",
        bash_command="echo $HOSTNAME",
    )

    bash_t1 >> bash_t2

from airflow import DAG
import pendulum
from airflow.operators.empty import EmptyOperator

with DAG(
        dag_id="dags_conn_test1",
        schedule=None,
        start_date=pendulum.datetime(2024, 6, 14, tz="Asia/Seoul"),
        catchup=False
) as dag:
    t1 = EmptyOperator(
        task_id="t1"
    )

    t2 = EmptyOperator(
        task_id="t2"
    )

    t3 = EmptyOperator(
        task_id="t3"
    )

    t4 = EmptyOperator(
        task_id="t4"
    )

    t5 = EmptyOperator(
        task_id="t5"
    )

    t6 = EmptyOperator(
        task_id="t6"
    )

    t7 = EmptyOperator(
        task_id="t7"
    )

    t8 = EmptyOperator(
        task_id="t8"
    )

    t1 >> t2
    t1 >> t3
    t3 >> t4
    t5 >> t4
    t4 >> t6
    t7 >> t6
    t6 >> t8

from airflow import DAG
import pendulum
from airflow.operators.empty import EmptyOperator

with DAG(
    dag_id="dags_conn_test2",
    schedule=None,
    start_date=pendulum.datetime(2024, 6, 14, tz="Asia/Seoul"),
    catchup=False
) as dag:
    
    t1 = EmptyOperator(
        task_id="t1"
    )

    t2 = EmptyOperator(
        task_id="t2"
    )

    t3 = EmptyOperator(
        task_id="t3"
    )

    t4 = EmptyOperator(
        task_id="t4"
    )

    t5 = EmptyOperator(
        task_id="t5"
    )

    t6 = EmptyOperator(
        task_id="t6"
    )

    t7 = EmptyOperator(
        task_id="t7"
    )

    t8 = EmptyOperator(
        task_id="t8"
    )

    t1 >> [t2, t3] >> t4
    t5 >> t4 
    [t4, t7] >> t6 >> t8

 

The scheduler does not appear to be running.

 

스케쥴러 실행이 안되는 경우 컨테이너를 조회하여 restart를 한다.

 

sudo docker container ps
sudo docker container restart {스케쥴러 컨테이너 ID}