[AIRFLOW] logical_date 의 개념 이해하기- execution_date 의 대체

과거에는 execution_date 라는 명칭으로 사용하던 개념이 있는데, 이게 execution 이라는 이름이 있어서 실제 실행된 시간의 개념으로 오해하는 문제가 있었다. (특히 crontab 에 익숙한 사람이라면 더더욱더)

 

이런 문제때문에 Airflow 2.2 부터는 execution_date 를 쓰지 않고, logical_date 를 쓰는 방향으로 변경되었다. (하위호환성을 위해 execution_date 도 쓸수 있긴하지만 쓰지말자 용어가 헛갈리게 하는 주범이다.)

https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-39+Richer+scheduler_interval

 

헛갈리는 개념? 논리적? 실행시간?

crontab 에서는 단순히 특정 스케쥴이 실행되기 때문에 실행시점과 스케쥴시점이 동일하다.

하지만, airflow 에서는 스케쥴되는 시점과 실행되는 시점이 다를수 있다.  즉, 2가지의 시간 관점이 존재한다.

 

• (1) 실제 실행된 시간 = task 가 서버에서 실행되는 실제시간 (즉, 재처리 하면 날짜가 바뀐다)
• (2) 스케쥴 간격에 따른 논리적인 시간 = logical_date (즉, 재처리해도 날짜는 유지되는 논리적인 값이다)

실제 실행된 시간

DAG 가 서버에서 실행된 시작과 종료시간을 의미한다. Airflow 실행이력에서 Started 와 Ended 값을 생각하면 된다. 

단순히 실행된 시간이기 때문에 재실행 하면 당연히 이 시간은 재처리된 시간으로 변경된다.

deprecate 된 execution_date 로 오해할 수 있는데, 여기서 말하는 실행된 시간의 개념과는 다르다.

execution_date 는 사실 logical_date 값이라고 봐야한다.

 

논리적 시간 = logical_date

스케쥴이 간격이 1시간이고, 시작시간이 2023년 1월 1일이라고 가정해보자.

0시 파티션의 데이터는 00:00~01:00 까지의 로그를 집계해야하며, 실제 데이터가 쌓이려면 01시 이후에 스케쥴이 실행되어야한다.

 

예를 들어, 아래와 같이 데이터를 만든다고 가정하면? 아래와 같은 local_date 와 data_interval 값이 맵핑된다.

이는 실제 실행되는 시점에 대한 값이 아니라, 논리적으로 다루려는 시간의 윈도우 범위를 의미하는 값이다.

시간	hdfs 경로	관련된 값
00:00 ~ 01:00	hdfs://my/user/foo/log/event/ymd=2023-01-01/hh24=00	logical_date : 2023-01-01 00:00 data_interval_start : 2023-01-01 00:00 data_interval_end : 2023-01-01 01:00
01:00 ~ 02:00	hdfs://my/user/foo/log/event/ymd=2023-01-01/hh24=01	logical_date : 2023-01-01 01:00 data_interval_start : 2023-01-01 01:00 data_interval_end : 2023-01-01 02:00
...	...	...

 

Airflow 의 실행이력 정보에서 Data interval start, Data interval end 의 값을 의미한다.

 

 

조금 더 알아보기

start_date 가 2023/01/01 00:00 이고, 스케쥴은 10분으로 정한 DAG 가 있다고 가정해보자. 코드로 표현하면 아래와 같다.

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from datetime import datetime, timedelta
import pendulum

local_tz = pendulum.timezone("Asia/Seoul")
default_args = {
    'start_date': datetime(2023, 1, 1, 0, tzinfo=local_tz),
}

with DAG(
        dag_id='sample_dag',
        default_args=default_args,
        schedule=timedelta(minutes=10),
        catchup=True
) as dag:
    EmptyOperator(task_id='end')

 

그렇다면, 아래와 같은 사이클로 스케쥴이 돌게 된다.  10분단위 스케쥴이므로 logical_date 의 시간이 "00:00" 인 사이클이 실제 invoke 되는 시점은 00:10 이 되며, 이때 큐상태나 워커상태에 따라 바로 실행될 수 있을수 있고 지연될 수 도 있는 특징이 존재한다.

사이클	logical_date data_interval_start	data_interval_end
1	2023/01/01 00:00	2023/01/01 00:10
2	2023/01/01 00:10	2023/01/01 00:20
3	2023/01/01 00:20	2023/01/01 00:30
...	...	...

 

schedule 의 timedelta 값은 텀블링 윈도우의 사이즈 개념으로 생각하면 이해가 좀더 쉽다.

그래서 만약, 기존 실행이력이 존재한 상태에서 timedelta 값이 바뀌게 되면 마지막 data_interval_end 값에 해당 timedelta 값을 더한값이 다음 data_interval_start 값이 된다. (물론 중간에 배치 사이클을 바꾸는게 흔치 않긴하지만)

글과 말로 열심히 써도 이해가 어려울텐데, 실제 샘플로직을 스케쥴링하면서 돌려보면 이해가 더 빠를것이니 한번 직접 돌려보도록 하자.