관리하는 데이터 파이프라인의 수가 늘어나면 이 중의 몇은 항상 실패하게 되며 어떻게 관리하느냐가 데이터 엔지니어의 삶에 큰 영향을 준다.
Incremental Update가 실패하면?
하루에 한번 동작하고 Incremental하게 업데이트하는 파이프라인이라면 ?
실패한 부분을 재실행하는 것이 얼마나 중요한가 ?
이제부터 할 이야기는 Incremental Update시에만 의미가 있다
- 다시 한번 가능하면 Full Refresh를 사용하는 것이 좋음
- 문제가 생겨도 다시 실행하면 됨
- Incremental Update는 효율성이 더 좋을 수 있지만 운영/유지보수의 난이도가 올라감
- 실수 등으로 데이터가 빠지는 일이 생길 수 있음
- 과거 데이터를 다시 다 읽어와야하는 경우 다시 모두 재실행을 해주어야함
Backfill의 용이성 여부 -> 데이터 엔지니어 삶에 직접적인 영향 !
- Backfill의 정의
- 실패한 데이터 파이프라인을 재실행 혹은 읽어온 데이터들의 문제로 다시 다 읽어와야 하는 경우를 의미
- Backfill 해결은 Incremental Update에서 복잡해짐
- Full Refresh에서는 간단. 그냥 다시 실행하면 됨
- 즉 실패한 데이터 파이프라인의 재실행이 얼마나 용이한 구조인가?
- 이게 잘 디자인된 것이 바로 Airflow
보통 Daily DAG를 작성한다고 하면 어떻게 할까 ?
- 지금 시간을 기준으로 어제 날짜를 계산하고 그 날짜에 해당하는 데이터를 읽어옴
from datetime import datetime, timedelta
# 지금 시간 기준으로 어제 날짜를 계산
y = datetime.now() - timedelta(1)
yesterday = datetime.strftime(y, '%Y-%m-%d')
# yesterday에 해당하는 데이터를 소스에서 읽어옴
# 예를 들어 프로덕션 DB의 특정 테이블에서 읽어온다면
sql = f"SELECT * FROM table WHERE DATE(ts) = '{yesterday}'"만약 이 코드가 실행되지 않으면 해당날짜의 incremental update는 빠져있을 것이다.
며칠 지난뒤에 해결하려하면 어려워짐. 코드를 해킹해야함(하드코딩)
실수를 하게 될 수도 있다. 복잡한 코든 어딜 건드려야할지도 모를 수 있고, 원상복구 시에도 실수할 수 있다.
DAG를 처음 만들 때부터 backfill이 쉽게가는 구조를 생각하지 않으면 몸이 고생한다.
Airflow는 이 문제를 시스템 차원에서 해결할 수 있도록 구성되어 있다.
어떻게 ETL을 구현해놓으면 이런일이 편해질까?
- 시스템적으로 이걸 쉽게 해주는 방법을 구현한다.
- 날짜별로 backfill 결과를 기록하고 성공 여부 기록: 나중에 결과를 쉽게 확인
- 이 날짜를 시스템에서 ETL의 인자로 제공
- 데이터 엔지니어는 읽어와야하는 데이터의 날짜를 계산하지 않고 시스템이 지정해준 날짜를 사용
- Airflow의 접근방식
- ETL별로 실행날짜와 결과를 메타데이터 데이터베이스에 기록
- 모든 DAG 실행에는 "execution_date"이 지정되어 있음
- execution_date으로 채워야하는 날짜와 시간이 넘어옴
- 이를 바탕으로 데이터를 갱신하도록 코드를 작성해야함
- 이점: backfill이 쉬워짐.
Daily Incremental Update를 구현해야 한다면 ?
- 예를 들어 2020년 11월 7일의 데이터부터 매일매일 하루치 데이터를 읽어온다고 가정
- 이 경우 언제부터 해당 ETL이 동작해야하나 ?
- 2020년 11월 8일
- 다르게 이야기하면 2020년 11월 8일날 동작하지만 읽어와야 하는 데이터의 날짜는 ?
- 2020년 11월 7일 : 이게 start_date가 됨
- 2020년 11월 7일 : 이게 start_date가 됨
- Airflow의 start_date은 시작 날짜라기보다는 처음 읽어와야하는 데이터의 날짜임
- execution_date는 읽어와야하는 데이터의 날짜로 설정됨
start_date와 execution_date 이해하기
- 2020-08-10 02:00:00로 start date로 설정된 daily job이 있다.
- catchup이 True로 설정되어 있다고 가정 (default=True)
- 지금 시간이 2020-08-13 20:00:00이고 처음으로 이 job이 활성화되었다
이 경우 이 job은 몇번 실행될까 ? (execution_date) 3번
- 2020-08-10 02:00:00 O
- 2020-08-11 02:00:00 X
- 2020-08-12 02:00:00 X
- 2020-08-13 02:00:00 X
Backfill과 관련된 Airflow 변수들
start_date : DAG가 처음 실행되는 날짜가 아니라 DAG가 처음 읽어와야 하는 데이터의 날짜/시간. 실제 첫 실행날짜는 start_date + DAG의 실행주기
execution_date : DAG가 읽어와야하는 데이터의 날짜와 시간
catchup : DAG가 처음 활성화된 시점이 start_date보다 미래라면 그 사이에 실행이 안된 것들을 어떻게 할것인지 결정해주는 파라미터. True가 디폴트값이고 이 경우 실행안된 것들을 모두 따라잡으려고 함. False가 되면 실행안된 것들을 무시함.
end_date : 이 값은 보통 필요하지 않으며 Backfill을 날짜 범위에 대해 하는 경우에만 필요
airflow dags backfill -s .... -e ....
