[5주차] Airflow BackFill(Incremental Update, execution_date)

BACKFILL

---

데이터파이프라인을 설계하고 운용할 때, 가끔 에러가 나거나, 이미 지난 날짜를 기준으로 데이터를 재처리 해야할 때가 있습니다. 이 때 백필은 재처리 작업을 의미합니다. 단어 의미 그대로 '매우는 작업'이라고 보시면 됩니다.

• Full Refrest를 한다면 backfill은 필요 없습니다.
• backfill은 일별 혹은 시간별 업데이트를 의미합니다.
  • 마지막 업데이트 시간 기준 backfill을 하는 경우라면(데이터 웨어하우스 기록 시간 기준), execution_date을 이용한 backfill은 필요하지 않습니다.
• 데이터의 크기가 커질수록 backfill 기능을 구현해두는 것은 필수입니다.
  • airflow는 이 backfill 작업을 굉장히 쉽게 만들지만, 데이터소스의 도움없인 불가능합니다.(created_date 등)

 

 

커맨드 라인을 활용한 BACKFILL

---

airflow dags backfill dag_id -s 2023-01-01 -e 2023-01-31

• -s : startdate
• -e : enddate
• 이 상황에만 사용 가능합니다.
  • catchUp : True
  • execution_date를 사용한 Incremental update의 구현
• start_date부터 시작하지만, end_date은 포함하지 않습니다
  
  • 위의 커맨드 라인에 의하면, 1월 1일부터 1월 30일까지 총 30개의 날짜를 BACKFILL합니다.
• 실행순서는 날짜/시간 순이 아닌 랜덤으로 진행됩니다. 만일 날짜 순으로 하고 싶다면 dag의 옵션에 이를 추가합니다.

default_args = {
 'depends_on_past': True
 }

 

 

실제 사용 예제

---

mysql_to_s3_nps = SqlToS3Operator(
    task_id = 'mysql_to_s3_nps',
    query = "SELECT * FROM prod.nps WHERE DATE(created_at) = DATE('{{ execution_date }}')",
    s3_bucket = s3_bucket,
    s3_key = s3_key,
    sql_conn_id = "mysql_conn_id",
    aws_conn_id = "aws_conn_id",
    verify = False,
    replace = True,
    pd_kwargs={"index": False, "header": False},    
    dag = dag
)

execution_date으로 incremental Update하는 Operator입니다. 해당 코드를 가지고 실행을 하면, 실행날짜에 맞춰서 실행되기 때문에 실행일에 맞춰서 데이터를 업데이트합니다. 저희는 2023년 1월 1일부터 1월 30일까지의 데이터를 Backfill 하기 위해서, CLI를 통해서 업데이트해보겠습니다.

dag = DAG(
    dag_id = 'MySQL_to_Redshift_v2',
    start_date = datetime(2022,12,30), # 날짜가 미래인 경우 실행이 안됨
    schedule = '0 9 * * *',  # 적당히 조절
    max_active_runs = 1,
    catchup = True,
    # default_args = {
    #     'retries': 1,
    #     'retry_delay': timedelta(minutes=3),
    # }
)

cachup은 True로 설정된 모습입니다. backfill을 위해서 catchup = True, incrementalUpdate를 위해 execution_date를 활용했기 때문에 backfill이 가능합니다.

airflow dags backfill MySQL_to_Redshift_v2 -s 2023-01-01 -e 2023-01-03

저는 1월 1일부터 1월 2일까지의 데이터를 backfill 하겠습니다.

실행모습

select * from tkddnr961224.nps order by created_at DESC;

 

정확히 1월 2일까지의 데이터를 잘 불러왔습니다.

 

그 외의 사례

---

Backfill은 특정 날짜의 데이터를 채우고 싶을 때도, 사용하지만 전 포스팅에서도 보여드렸던 것처럼 에러가 난 특정 task만 backfill 하는 방법도 있습니다. 이는 웹 UI를 통해서 원클릭으로 해결이 되며, 그보다 에러 메시지를 파악하고, 에러를 해결한 뒤에 웹 ui를 통해 backfill 합니다.