11.Backfill과 Airflow

Incremental Update의 실패

<aside> 💡 관리하는 데이터 파이프라인의 수가 늘어나면 이 중의 몇은 항상 실패하게 되며 이를 어떻게 관리하느냐가 데이터 엔지니어의 삶에 큰 영향을 준다

</aside>

Incremental Update가 실패하면?
하루에 한번 동작하고 Incremental하게 업데이트하는 파이프라인이라면?
1. 만약 아래와 같이 실행될 경우, 이틀간의 정보가 빠져있게 된다
실패한 부분을 재실행하는 것이 얼마나 중요한가?

⇒ 관리하는 데이터 파이프라인의 수가 늘어나면 이 중의 몇은 항상 실패하게 되며 이를 어떻게 관리하느냐가 데이터 엔지니어의 삶에 큰 영향을 준다

이제부터 할 이야기는 Incremental Update시에만 의미가 있습니다.
- 다시 한번 가능하면 Full Refresh를 사용하는 것이 좋음
  - 문제가 생겨도 다시 실행하면 됨
- Incremental Update는 효율성이 더 좋을 수 있지만 운영/유지보수의 난이도가 올라갑니다.
  - 실수등으로 데이터가 빠지는 일이 생길 수 있음
  - 과거 데이터를 다시 다 읽어와야하는 경우 다시 모두 재실행을 해주어야함

Backfill

Backfill의 용이성 여부 → 데이터 엔지니어 삶에 직접적인 영향!

Backfill의 정의

<aside> 💡 실패한 데이터 파이프라인을 재실행 혹은 읽어온 데이터들의 문제로 다시 다 읽어와야하는 경우를 의미

</aside>
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}'"

그런데 지난 1년치 데이터를 Backfill 해야한다면?
```
from datetime import datetime, timedelta
y = datetime.now() - timedelta(1)
yesterday = datetime.strftime(y, '%Y-%m-%d')
yesterday = '2023-01-01'
# yesterday에 해당하는 데이터를 소스에서 읽어옴
# 예를 들어 프로덕션 DB의 특정 테이블에서 읽어온다면
sql = f"SELECT * FROM table WHERE DATE(ts) = '{yesterday}'"
```
- 기존 ETL 코드를 조금 수정(반복문을 추가)해서 지난 1년치 데이터에 대해 돌린다
- 실수하기 쉽고 수정하는데 시간이 걸림
- 읽어와야 할 데이터를 현재 시간 기준으로 계산하는 순간, 빠지는 데이터가 생겨 나중에 Backfill 작업을 수행하려고 하면 오류가 발생한다.

어떻게 ETL을 구현해놓으면 이런 일이 편해질까?

시스템적으로 이걸 쉽게 해주는 방법을 구현한다
- Airflow는 Incremental update 혹은 Full Refresh를 구분하지 못한다.
  - 모든 DAG가 incremental 하게 작동된다고 가정함
- 날짜 별로 backfill 결과를 기록하고 성공 여부 기록: 나중에 결과를 쉽게 확인
- 이 날짜를 시스템에서 ETL의 인자로 제공
- 데이터 엔지니어는 읽어와야하는 데이터의 날짜를 계산하지 않고 시스템(Airflow)이 지정해준 날짜를 사용
  - execution_date
Airflow의 접근방식
- ETL 별로 실행날짜와 결과를 메타데이터 데이터베이스에 기록
- 모든 DAG 실행에는 “execution_date”이 지정되어 있음
  - execution_date으로 채워야하는 날짜와 시간이 넘어옴
- 이를 바탕으로 데이터를 갱신하도록 코드를 작성해야함
- 이점: backfill이 쉬워짐