[TIL 6/8] Airflow Day 4

PK 유지 방법

여기에서 기록했던 것처럼, 빅데이터 웨어하우스에서는 PK가 무시된다. 오늘은 PK를 유지하는 나름의 방법을 배웠다. 아래는 예시다.

1. 임시 테이블을 만든다. 원래 테이블의 내용을 임시 테이블 t에 복사한다.

CREATE TEMP TABLE t AS SELECT * FROM heering.weather_forecast;

2. DAG는 임시 테이블(staging table)에 레코드를 추가한다. 중복 데이터가 들어갈 수 있다.

3. 원래 테이블 내용 삭제한다.

DELETE FROM heering.weather_forecast;

4. 중복을 없앤 형태로 새로운 테이블을 생성한다.

INSERT INTO heering.weather_forecast
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; -- 일련번호인 seq가 1인 것만.

5. 매번 이렇게 새로 덮어쓰는 형식의 업데이트. autocommit=True라면, 최소 3번과 4번은 transaction으로 처리되어야 함.

Backfill 관련 Airflow 변수

이름	설명
start_date	처음 읽어와야 하는 데이터의 날짜. ETL의 첫 동작은 기록 시작일인 start_date + DAG의 실행주기부터.
execution_date	읽어와야 하는 데이터의 날짜로 설정됨. 시스템의 변수로 읽어와야 하는 데이터의 날짜를 지정
catchup	⚠ default가 True이므로 주의, DAG가 처음 활성화된 시점이 start_date보다 미래라면 그 사이에 실행 안된 걸 실행함.
end_date	보통 불필요, Backfill을 날짜 범위에 대해 하는 경우에만