Airflow ETL에서 Primary Key Uniqueness SQL로 보장하기 - TIL230608

📚 KDT WEEK 10 DAY 4 TIL

• Primary Key Uniqueness 보장하기

 

---

 

🟥 Primary Key Uniqueness 보장하기

• PK : 테이블에서 하나의 레코드를 유일하게 지칭할 수 있는 필드(들)
• 관계형 데이터베이스 시스템이 Primary key의 값이 중복 존재하는 것을 막아준다.
• 빅데이터 기반 데이터 웨어하우스들은 Primary key Uniqueness를 보장하지 않으므로 엔지니어가 보장해야함

날씨 데이터 API(https://openweathermap.org/api)를 가져오는 과정에서 PK Uniquness를 보장해보려고 한다.

SQL 문법을 사용해서 Upsert(데이터가 있는 경우 새로 업데이트)방식을 사용할 것이다.

참고로 적재 대상 DB인 Redshift에서는 Upsert 기능을 제공한다고 한다. 이건 단순 학습용으로 기록되는 글이다!

 

 

다음의 날씨 정보 테이블을 사용한다고 가정해보자.

 

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

 

• date : 해당 날짜(의 날씨 정보)
• temp : 기온
• min_temp : 최저 기온
• max_temp : 최대 기온
• created_date : 날씨 정보 생성 날짜 > 데이터가 들어가면서 현재 시간 자동 입력

날씨 정보의 경우 해당 날짜에 가까워질수록 더 정확한 정보가 되므로 created_date가 최신일수록 가치있는 데이터로 판단한다.

 

날씨 정보를 어떤 API를 통해 가져와서 Airflow ETL을 Incremental Update(새로운 데이터 삽입) 방식을 사용한다면,

2일차 3일차 데이터에는 같은 날짜의 중복 날씨 데이터가 생길 것이다.

이때 데이터를 적재하는 과정에서 중복 데이터 중 더 가치있는 데이터를 우선하여 적재해보자.

 

Airflow에서 Redshift를 연결하거나, ETL을 만드는 과정은 이전 글에 적었으니 모두 생략하였다.

 

데이터 중 같은 date 필드를 묶어 created_date를 정렬하려면, ROW_NUMBER() 함수를 사용하면 된다.

PARTITION BY에 date 필드를 넣어 묶고, ORDER BY에 created_date를 넣고 내림차순(DESC)를 적용한다.

ROW_NUMBER() OVER (PARTITION BY date ORDER BY created_date DESC) seq

 

해당 함수를 사용해서 SELECT 문으로 출력해보면 다음과 같은 형태로 나올 것이다!

 

date	created_date	temp	seq
2021-01-01	2021-01-03	6	1
2021-01-01	2021-01-02	5	2
2021-01-01	2021-01-01	2	3
2021-01-02	2021-01-03	3	1
2021-01-02	2021-01-01	4	2

 

이 부분에서 WHERE을 통해 seq = 1인 부분만 데이터를 적재하면 될 것이다.

따라서 적재시에 사용할 INSERT 문을 다음과 같이 사용한다.

 

INSERT INTO {schema}.{table}
SELECT date, temp, min_temp, max_temp FROM (
	SELECT *, ROW_NUMBER() OVER (PARTITION BY date ORDER BY created_date DESC) seq
	FROM t
)
WHERE seq = 1;

 

데이터 적재시 INSERT문을 위처럼 사용하면 중복데이터는 새로 업데이트하고, 새로운 데이터는 새로 적재하게 된다.

 

전체 코드

 

from airflow import DAG
from airflow.decorators import task
from airflow.models import Variable
from airflow.providers.postgres.hooks.postgres import PostgresHook

from datetime import datetime
from datetime import timedelta

import requests
import logging
import json

# Airflow에 저장된 정보를 통해 Redshift 연결
def get_Redshift_connection():
    # autocommit is False by default
    hook = PostgresHook(postgres_conn_id='redshift_dev_db')
    return hook.get_conn().cursor()

# ETL (use Airflow Task Decorator)
@task
def etl(schema, table, lat, lon, api_key):

    # 날씨 API 데이터 요청
    url = f"https://api.openweathermap.org/data/2.5/onecall?lat={lat}&lon={lon}&appid={api_key}&units=metric&exclude=current,minutely,hourly,alerts"
    response = requests.get(url)
    data = json.loads(response.text)

    """
    {'dt': 1622948400, 'sunrise': 1622923873, 'sunset': 1622976631, 'moonrise': 1622915520, 'moonset': 1622962620, 'moon_phase': 0.87, 'temp': {'day': 26.59, 'min': 15.67, 'max': 28.11, 'night': 22.68, 'eve': 26.29, 'morn': 15.67}, 'feels_like': {'day': 26.59, 'night': 22.2, 'eve': 26.29, 'morn': 15.36}, 'pressure': 1003, 'humidity': 30, 'dew_point': 7.56, 'wind_speed': 4.05, 'wind_deg': 250, 'wind_gust': 9.2, 'weather': [{'id': 802, 'main': 'Clouds', 'description': 'scattered clouds', 'icon': '03d'}], 'clouds': 44, 'pop': 0, 'uvi': 3}
    """
    
    # 받은 요청의 데이터에 따라 필요한 필드만 저장
    ret = []
    for d in data["daily"]:
        day = datetime.fromtimestamp(d["dt"]).strftime('%Y-%m-%d')
        ret.append("('{}',{},{},{})".format(day, d["temp"]["day"], d["temp"]["min"], d["temp"]["max"]))

    cur = get_Redshift_connection()
    
    # 원본 테이블이 없다면 생성
    create_table_sql = f"""CREATE TABLE IF NOT EXISTS {schema}.{table} (
    date date,
    temp float,
    min_temp float,
    max_temp float,
    created_date timestamp default GETDATE()
);"""
    logging.info(create_table_sql)

    # 임시 테이블 생성 (CTAS 원본테이블)
    create_t_sql = f"""CREATE TEMP TABLE t AS SELECT * FROM {schema}.{table};"""
    logging.info(create_t_sql)
    try:
        cur.execute(create_table_sql)
        cur.execute(create_t_sql)
        cur.execute("COMMIT;")
    except Exception as e:
        cur.execute("ROLLBACK;")
        raise

    # 임시 테이블 데이터 입력
    insert_sql = f"INSERT INTO t VALUES " + ",".join(ret)
    logging.info(insert_sql)
    try:
        cur.execute(insert_sql)
        cur.execute("COMMIT;")
    except Exception as e:
        cur.execute("ROLLBACK;")
        raise

    # 기존 테이블 대체
    alter_sql = f"""DELETE FROM {schema}.{table};
      INSERT INTO {schema}.{table}
      SELECT date, temp, min_temp, max_temp FROM (
        SELECT *, ROW_NUMBER() OVER (PARTITION BY date ORDER BY created_date DESC) seq
        FROM t
      )
      WHERE seq = 1;"""
    logging.info(alter_sql)
    try:
        cur.execute(alter_sql)
        cur.execute("COMMIT;")
    except Exception as e:
        cur.execute("ROLLBACK;")
        raise

# DAG 정의
with DAG(
    dag_id = 'Weather_to_Redshift_v2',
    start_date = datetime(2022,8,24), # 날짜가 미래인 경우 실행이 안됨
    schedule = '0 4 * * *',  # 매일 4시마다
    max_active_runs = 1,
    catchup = False,
    default_args = {
        'retries': 1,
        'retry_delay': timedelta(minutes=3),
    }
) as dag:

	# 순서대로 스키마이름, 테이블이름, 서울좌표1, 서울좌표2, 날씨API KEY
    etl("sunhokim_public", "weather_forecast_v2", 37.5665, 126.9780, Variable.get("open_weather_api_key"))

---