[8주차] 데이터 파이프라인과 Airflow (4)

🙂 Primary Key Uniqueness

• 테이블에서 하나의 레코드를 유일하게 지칭할 수 있는 필드
  • 하나의 필드가 일반적이지만 다수의 필드를 사용할 수도 있음
  • CREATE TABLE 사용시 지정
• 관계형 DBMS가 PK의 값이 중복 존재하는 것을 방지

 

✔ 빅데이터 기반 DW는 PK보장 X

데이터 인력을 통해 보장

보장하는 데 메모리와 시간이 더 들기 때문에 대용량 데이터 적재가 걸림돌

🍦 PK 유지 방법 1

• 임시 테이블을 만들고 현재 모든 레코드를 복사
• 임시 테이블에 새로 데이터소스에서 읽어들인 레코드를 복사
• 중복을 걸러주는 SQL 작성
  • 최신 레코드를 우선 순위로 선택
  • ROW_NUMBER를 이용해서 PK로 파티션을 잡고 ts 필드르 order desc를 수행해 PK별로 레코드를 선택
• 위의 SQL을 바탕으로 최종 원본 테이블로 복사
  • 원본 테이블에서 레코드를 삭제
  • 임시 테이블을 원본 테이블로 복사

🍦 PK 유지 방법 2

• CREATE TEMP TABLE t AS SELECT * FROM [원본테이블];
• DAG를 통해 임시 테이블로 레코드 추가
• DELETE FROM [원본테이블];
• 중복을 없앤 형태로 새로운 테이블 생성
• INSERT INTO [원본테이블]

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

✔ Upsert

• PK를 기준으로 존재하는 레코드라면 새 정보로 수정
• 존재하지 않는 레코드라면 새 레코드로 적재
• 보통 DW마다 Upsert를 효율적으로 해주는 문법 지원

https://michaeljswart.com/

---

🙂 Backfill과 Airflow

Incremental Update가 실패하면?

✔ 정의

• 실패한 데이터 파이프라인을 재실행 혹은 읽어온 데이터의 문제로 다시 다 읽어와야하는 경우를 의미
• Backfill 해결은 Incremental Update에서 복잡해짐 (Full Refresh에서는 재실행으로 해결)

🍦 Daily Incremental Update를 구현해야 한다면?

• 2020-11-07의 데이터부터 매일 하루치 데이터를 읽어온다고 가정
• ETL이 언제부터 동작해야하나? -> 2020-11-08
• 2020-11-08 동작하는 DAG가 읽어와야 하는 데이터 날짜는? -> 2020-11-07 (start_date)
• Airflow의 start_date는 시작 날짜라기 보다는 처음 읽어와야하는 데이터의 날짜
• execution_date는 읽어와야하는 데이터의 날짜

🍦 start_date와 execution_date

• daily job의 start_date = 2020-08-10 02:00:00 (catchup = True)
• 지금 시간이 2020-08-13 20:00:00이고 처음으로 이 job이 활성화
• 이 경우 job은 몇번 실행될까? (execution_date)
  • 2020-08-10 02:00:00
  • 2020-08-11 02:00:00
  • 2020-08-12 02:00:00
  • 2020-08-13 02:00:00

오늘 도는 배치가 있습니다. 오늘자 데이터는 집계해도 의미가 없어요. 오늘이 아직 안끝났으니까요.
오늘자 데이터 집계는 내일 해야 합니다.

그럼 오늘 도는 배치는? 보통 어제자 데이터를 집계하는 배치입니다.
여기서 execution_date는 어제를 가리킵니다.
배치가 시작한것은 오늘이지만, 실행을 기준으로 하는 날짜를 어제로 잡은것이죠.
여기서 주의하실 점은 execution_date가 무조건 하루 전을 가리키는것은 아니라는 점입니다.
만약 1달에 1번 도는 배치가 오늘 돈다면, execution_date는 한달 전을 가리킵니다.
한달에 한번 도는 배치는 한달동안의 데이터를 집계할테니, 한달 전을 실행 기준일자로 잡는것이죠.
무조건 하루전을 가리키는게 아니란걸 유의해야합니다!
한시간 전일수도 있고, 하루 전일수도 있고 1년전일수도 있어요.
이건 배치 주기에 따라서 달라집니다.

 

✔ Backfill과 관련된 Airflow 변수

• start_date: DAG가 처음 실행되는 날짜 X -> DAG가 처음 읽어와야하는 데이터의 날짜/시간.
                    실제 첫 실행날짜는 start_date + DAG 실행주기
• execution_date: DAG가 읽어와야하는 데이터의 날짜와 시간
• catchup: DAG가 처음 활성화된 시점이 start_date 보다 미래라면 그 사이에 실행이 안된 것을 어떻게 할 것인지 판단
• end_date: Backfill을 날짜 범위에 대해 하는 경우에만 필요