Apache Airflow 재수행 방법 정리

Apache Airflow을 구축해서 운영하다보면 이런저런 이유들로 인해 DAG을 재수행해야하는 일이 발생한다.

raw 데이터 오류, 집계 로직 오류 등의 이유로..

 

재수행하는 방법에 대해서 정리해본다.

일단 크게 DAG run 단위로 재수행하거나 task 단위로 재수행하는 방법으로 구분할 수 있을 것 같다.

 

# DAG run clear 하기

DAG run을 clear해서 DAG run이 다시 수행되게 하는 방법이다.

DAG run을 눌러 나오는 Clear 버튼을 눌러 clear하면 해당 DAG run이 재수행된다.

상단의 Browse > DAG Runs 메뉴에서 목록으로 DAG run을 조회하여 Clear the status 버튼을 통해 Clear하는 방법도 가능하다.

 

# Task clear 하기

task를 clear해서 task가 다시 수행되게 하는 방법이다.

UI에서 clear 하기

task를 눌러서 나오는 팝업창에서 Clear 버튼을 눌러서 삭제할 수 있다.

버튼 옆에 보면 몇 가지 옵션과 같이 선택할 수 있는데 다음의 의미를 가지므로 잘 알아두면 유용하게 사용할 수 있다.

• Past : 해당 task의 과거 시점 task들을 같이 삭제한다. 많은 task들이 삭제될 수 있으므로 주의해야 한다.
• Future : 해당 task의 미래 시점 task들을 같이 삭제한다. 많은 task들이 삭제될 수 있으므로 주의해야 한다.
• Upstream : 해당 task의 의존성 상 상위 task들을 같이 삭제한다.
• Downstream : 해당 task의 의존성 상 하위 task들을 같이 삭제한다.
• Recursive : 부모 DAG, 하위 DAG의 task들을 같이 삭제하는 것으로 보이는데, sub DAG을 사용한 경우 해당되는 것 같다. (이 옵션은 직접 사용해보지 않아서 정확하지 않다.)
• Failed : 실패한 Task만 삭제한다.

여러개 옵션을 동시에 선택할 수 있기 때문에 조합해서 사용할 수 있고, 예를 들어 Past, Failed 옵션을 선택한다면 과거에 실패했던 task들을 삭제할 수 있다.

서로 의존성이 없는 관계의 task는 Upstream, Downstream 옵션과 함께 수행되더라도 삭제되지 않는다.

 

CLI에서 clear 하기

task를 삭제하는 Airflow CLI 명령어로 삭제하는 것도 가능한데, CLI에서 해야하는 불편함이 있지만 UI에서는 불가능한 조건을 처리할 수 있다.

가이드 문서 : https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#clear

airflow tasks clear [-h] [-R] [-d] [-e END_DATE] [-X] [-x] [-f] [-r]
                    [-s START_DATE] [-S SUBDIR] [-t TASK_REGEX] [-u] [-y]
                    dag_id

기본적으로 파라미터로 전달한 dag_id의 모든 task들을 삭제한다.

파라미터 설명

• dag_id
  • The id of the dag
• -R, --dag-regex
  • Search dag_id as regex instead of exact string
  • Default: False
  • 설정할 경우 dag_id 파라미터를 정규표현식으로 처리한다. 여러 dag에 대한 처리를 한번에 할 때 유용하다.
• -d, --downstream
  • Include downstream tasks
  • Default: False
• -e, --end-date
  • Override end_date YYYY-MM-DD
  • 재수행할 data_interval(aka. execution_date) 범위 중 끝 시간을 지정한다.
  • 필수 파라미터이다.
  • 가이드 문서에서는 YYYY-MM-DD 형식으로만 지정할 수 있어 보이지만 2021-12-01T00:00:00+09:00 형식으로 타임존 포함하여 지정할 수 있다.
    Airflow의 기본 타임존이 UTC이기 때문에 2021-12-01 이렇게만 지정하면 2021-12-01T00:00:00+00:00 으로 입력한 것으로 처리되기 때문에 한국 시간으로는 2021-12-01T09:00:00+09:00 가 된다.
    때문에 한국 시간 기준으로 입력할 때 타임존 정보까지 입력하면 편하다.
• -X, --exclude-parentdag
  • Exclude ParentDAGS if the task cleared is a part of a SubDAG
  • Default: False
• -x, --exclude-subdags
  • Exclude subdags
  • Default: False
• -f, --only-failed
  • Only failed jobs
  • Default: False
• -r, --only-running
  • Only running jobs
  • Default: False
• -s, --start-date
  • Override start_date YYYY-MM-DD
  • 재수행할 data_interval(aka. execution_date) 범위 중 시작 시간을 지정한다.
  • 필수 파라미터이다.
  • 가이드 문서에서는 YYYY-MM-DD 형식으로만 지정할 수 있어 보이지만 2021-12-01T00:00:00+09:00 형식으로 타임존 포함하여 지정할 수 있다.
    Airflow의 기본 타임존이 UTC이기 때문에 2021-12-01 이렇게만 지정하면 2021-12-01T00:00:00+00:00 으로 입력한 것으로 처리되기 때문에 한국 시간으로는 2021-12-01T09:00:00+09:00 가 된다.
    때문에 한국 시간 기준으로 입력할 때 타임존 정보까지 입력하면 편하다.
    
• -S, --subdir
  • File location or directory from which to look for the dag. Defaults to '[AIRFLOW_HOME]/dags' where [AIRFLOW_HOME] is the value you set for 'AIRFLOW_HOME' config you set in 'airflow.cfg'
  • Default: "[AIRFLOW_HOME]/dags"
• -t, --task-regex
  • The regex to filter specific task_ids to backfill (optional)
  • 삭제할 task의 task_id를 정규 표현식으로 지정할 수 있다.
  • 전체 task가 아니라 일부 task만 삭제하고자 할 때 사용할 수 있다.
• -u, --upstream
  • Include upstream tasks
  • Default: False
• -y, --yes
  • Do not prompt to confirm reset. Use with care!
  • Default: False
  • 명령어를 실행하면 삭제할 task 목록들을 출력하고 한번 컨펌을 받는 절차가 있는데 이 파라미터를 지정하면 그 절차가 생략된다.

# Backfill로 재수행 하기

Backfill이 수행되지 않은 기간의 DAG run들을 수행해서 채우는데 사용되지만, --reset-dagruns 파라미터를 지정함으로써 재수행도 할 수 있다.

Backfill은 기본적으로는 dag 단위를 실행하기 때문에 task 단위로 실행하려면 --task-regex 파라미터를 사용해야 한다.

가이드 문서 : https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#backfill

airflow dags backfill [-h] [-c CONF] [--delay-on-limit DELAY_ON_LIMIT] [-x]
                      [-n] [-e END_DATE] [-i] [-I] [-l] [-m] [--pool POOL]
                      [--rerun-failed-tasks] [--reset-dagruns] [-B]
                      [-s START_DATE] [-S SUBDIR] [-t TASK_REGEX] [-v] [-y]
                      dag_id

파라미터 설명

• dag_id
  • The id of the dag
• -c, --conf
  • JSON string that gets pickled into the DagRun's conf attribute
• --delay-on-limit
  • Amount of time in seconds to wait when the limit on maximum active dag runs (max_active_runs) has been reached before trying to execute a dag run again
  • Default: 1.0
• -x, --donot-pickle
  • Do not attempt to pickle the DAG object to send over to the workers, just tell the workers to run their version of the code
  • Default: False
• -n, --dry-run
  • Perform a dry run for each task. Only renders Template Fields for each task, nothing else
  • Default: False
• -e, --end-date
  • Override end_date YYYY-MM-DD
  • 실행할 data_interval(aka. execution_date) 범위 중 끝 시간을 지정한다.
  • 필수 파라미터이다.
  • 가이드 문서에서는 YYYY-MM-DD 형식으로만 지정할 수 있어 보이지만 2021-12-01T00:00:00+09:00 형식으로 타임존 포함하여 지정할 수 있다.
    Airflow의 기본 타임존이 UTC이기 때문에 2021-12-01 이렇게만 지정하면 2021-12-01T00:00:00+00:00 으로 입력한 것으로 처리되기 때문에 한국 시간으로는 2021-12-01T09:00:00+09:00 가 된다.
    때문에 한국 시간 기준으로 입력할 때 타임존 정보까지 입력하면 편하다.
    
• -i, --ignore-dependencies
  • Skip upstream tasks, run only the tasks matching the regexp. Only works in conjunction with task_regex
  • Default: False
  • backfill이 기본적으로 dag 단위로 수행되는데, task_regex 파라미터로 특정 task만 실행할 때 해당 task의 의존성 상 상위 task들을 제외하고 실행하는 것이다. 디폴트는 해당 task의 의존성 상 상위 task가 있다면 먼저 수행하지만 굳이 다시 수행할 필요가 없다면 제외하고 실행하고 싶을 수 있다.
• -I, --ignore-first-depends-on-past
  • Ignores depends_on_past dependencies for the first set of tasks only (subsequent executions in the backfill DO respect depends_on_past)
  • Default: False
• -l, --local
  • Run the task using the LocalExecutor
  • Default: False
• -m, --mark-success
  • Mark jobs as succeeded without running them
  • Default: False
  • 실제로 dag을 실행하지는 않고 단지 success 상태로 설정만 한다.
• --pool
  • Resource pool to use
• --rerun-failed-tasks
  • if set, the backfill will auto-rerun all the failed tasks for the backfill date range instead of throwing exceptions
  • Default: False
• --reset-dagruns
  • if set, the backfill will delete existing backfill-related DAG runs and start anew with fresh, running DAG runs
  • Default: False
  • 이미 data_interval(aka. execution_date)에 실행되었던 DAG run이 존재하더라도 그것을 삭제하고 다시 실행한다.
• -B, --run-backwards
  • if set, the backfill will run tasks from the most recent day first. if there are tasks that depend_on_past this option will throw an exception
  • Default: False
  • 원래 오래전 시간대부터 dag이 실행되지만 지정할 경우 최근 시간대부터 거꾸로 실행된다. dag 정의 시 인자 중 depend_on_past를 지정했다면 과거 시간대부터 실행되어야하기 때문에 이 파라미터를 지정하면 예외가 발생한다.
• -s, --start-date
  • Override start_date YYYY-MM-DD
  • 실행할 data_interval(aka. execution_date) 범위 중 시작 시간을 지정한다.
  • 필수 파라미터이다.
  • 가이드 문서에서는 YYYY-MM-DD 형식으로만 지정할 수 있어 보이지만 2021-12-01T00:00:00+09:00 형식으로 타임존 포함하여 지정할 수 있다.
    Airflow의 기본 타임존이 UTC이기 때문에 2021-12-01 이렇게만 지정하면 2021-12-01T00:00:00+00:00 으로 입력한 것으로 처리되기 때문에 한국 시간으로는 2021-12-01T09:00:00+09:00 가 된다.
    때문에 한국 시간 기준으로 입력할 때 타임존 정보까지 입력하면 편하다.
    
• -S, --subdir
  • File location or directory from which to look for the dag. Defaults to '[AIRFLOW_HOME]/dags' where [AIRFLOW_HOME] is the value you set for 'AIRFLOW_HOME' config you set in 'airflow.cfg'
  • Default: "[AIRFLOW_HOME]/dags"
• -t, --task-regex
  • The regex to filter specific task_ids to backfill (optional)
  • 실행할 task의 task_id를 정규 표현식으로 지정할 수 있다.
  • 전체 task가 아니라 일부 task만 실행하고자 할 때 사용할 수 있다.
• -v, --verbose
  • Make logging output more verbose
  • Default: False
• -y, --yes
  • Do not prompt to confirm reset. Use with care!
  • Default: False
  • 명령어를 실행하면 삭제할 task 목록들을 출력하고 한번 컨펌을 받는 절차가 있는데 이 파라미터를 지정하면 그 절차가 생략된다.

 

# 재수행 방법 정리

각 방법 별 재수행 가능 여부를 정리하면 다음과 같다.

CLI 명령어 중에 dag run clear 하는 명령어는 존재하지 않지만, dag의 모든 task를 삭제하면 dag run clear한 것과 유사하기 때문에 가능한 것과 마찬가지인 것 같다. (실제로 동일한 것일 수도..)

재수행 대상 \ 방법	[UI] clear	[CLI] airflow tasks clear	[CLI] backfill --reset-dagruns
dag run	가능	가능 (전체 task clear)	가능
task	가능	가능	가능 (--task-regex)

 

# 상황별 재수행 방법

schedule_interval 없는 DAG

backfill 명령은 schedule_interval이 설정된 DAG만 수행할 수 있다.

TriggerDagRunOperator로 실행되는 DAG이라면 upstream DAG의 trigger task를 backfill하거나 task clear함으로써 재수행할 수 있다.

만약 그냥 수동으로만 실행하는 DAG이고 clear할 dag run, task가 없는데 특정 기간에 대해 실행하고 싶다면,

trigger CLI를 여러번 실행하는 방법 밖에 없을 것 같다.

https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#trigger