[TIL — 30일차] 데이터 엔지니어링: 데이터 파이프라인, Airflow (4)

airflow.cfg

Airflow의 설정 파일을 관리하는 cfg 파일

docker-compose를 사용하여 배포 및 실행 시 Airflow_Home 에 덮어쓰기 되어 적용할 수 있음

  volumes:
    - ${AIRFLOW_PROJ_DIR:-.}/data:/opt/airflow/data
    - ${AIRFLOW_PROJ_DIR:-.}/dags:/opt/airflow/dags
    - ${AIRFLOW_PROJ_DIR:-.}/logs:/opt/airflow/logs
    - ${AIRFLOW_PROJ_DIR:-.}/config:/opt/airflow/config
    - ${AIRFLOW_PROJ_DIR:-.}/plugins:/opt/airflow/plugins
    - ./airflow.cfg:/opt/airflow/airflow.cfg # 덮어쓰기

환경 설정 파일이 수정되었다면 이를 실제로 반영하기 위해서 webserver. scheduler를 restart 해야함

Variable

Airflow의 Variable은 UI외 Config로 설정할 수 있음. 단 Config로 설정 시 UI에서는 보이지 않음

• AIRFLOW_VAR_{key}: {value}

Metadata DB의 암호화

fernet_key를 airflow.cfg에 설정하여 DB 내용을 암호화하여 저장할 수 있음

Airflow와 타임존

airflow.cfg에는 두 종류의 타임존 관련 키가 존재

1. default_timezone
2. default_ui_timezone

• start_date, end_date, schedule: default_timezone에 지정된 타임존을 따름
• execution_date와 로그 시간: 항상 UTC를 따름

가장 좋은 방법은 UTC를 일관되게 사용하는 것

dags 폴더에서 코딩시 작성한다면 주의할 점

Airflow는 dags 폴더를 주기적으로 스캔하기 때문에 top Level에 선언된 코드가 있다면 스캔 주기에 따라 계속 실행될 수 있음

[embed]Airflow Best Practices Airflow의 공식 문서의 Best Practices를 번역 및 요약한 내용입니다.medium.com

# /airflow.cfg

[core]
dags_folder = /var/lib/airflow/dags
# How often (in seconds) to scan the DAGs directory for new files. Default to 5 minutes.
dag_dir_list_interval = 300

Primary Key Uniqueness 보장하기

일반적으로 데이터 웨어하우스에서는 Primary Key 값이 중복되는 것을 막아주는 역할을 하지 않음

그러나 Incremental Update를 진행하다 보면 중복되는 경우가 생길 수 있기 때문에 데이터 웨어하우스를 관리하면서 PK Uniqueness를 보장할 수 있는 방법을 고려해야 함

• Primary Key Uniqueness: 테이블에서 하나의 레코드를 유일하게 지칭할 수 있는 필드

빅데이터 기반 데이터 웨어하우스가 PK Uniqueness를 지키지 않는 이유

• PK Uniqueness를 보장하는데 메모리와 시간이 들기 때문
• PK Uniqueness를 보장하는 것은 데이터 인력의 책임

created_date 활용하기

데이터를 증분 저장할 때 Created_date를 사용하여 동일한 Key를 가진 데이터의 경우 최신 데이터로 덮어쓰도록 구현한다

ROW_NUMBER를 사용하여 간단하게 UPSERT를 구현할 수 있음

CREATE TABLE keeyong.weather_forecast (
date date primary key,
temp float,
min_temp float,
max_temp float,
created_date timestamp default GETDATE()
);

SELECT date, temp, min_temp, max_temp, created_date
FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY date ORDER BY created_date DESC) seq
FROM t
)
WHERE seq = 1;

중복을 없앤 형태로 새로운 테이블을 생성

Upsert란?

Primary Key를 기준으로 존재하는 레코드라면 새 정보로 수정, 존재하지 않는 레코드라면 새 레코드로 적재

Backfill과 Airflow

Full Refresh를 사용하면, 문제가 생길 경우 다시 실행하면 되므로 가능하다면 이 방법을 사용해야 함 그러나 Incremental Update를 사용할 경우 효율성은 더 좋지만, 운영 / 유지보수 난이도가 올라감

Backfill의 정의

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

ex.) 2025–11–09일 파이프라인의 오류로 11–09일 데이터만 저장이 안 된 상태

Backfill과 관련된 Airflow 변수

[embed]Airflow 스케줄링과 Execution Date 동작 원리 Apache Airflow 스케줄링 시 Execution Date 설정할 때 스케줄되어 실행되는 DAG와 backfill, trigger로 실행되는 DAG에 설정되는 Execution_date 차이를 정리medium.com

• logical_date(execution_date): 시스템 변수로 읽어와야 하는 데이터의 날짜를 지정 (schduled되어 실행된 DAG 경우 date_interval_start와 동일)
• date_interval_start: 데이터를 읽어와야 하는 데이터의 기간의 시작 지점
• date_interval_end: 데이터를 읽어와야 하는 기간의 종료 지점