DBT의 추가 기능 알아보기 (seed, sources, snapshots, tests) - TIL230623

📚 KDT WEEK 12 DAY 5 TIL

• Dbt Seed
• Dbt Sources
• Dbt Snapshots
• Dbt tests

 

---

 

🟥 DBT Seed

 

많은 dimension 테이블들은 크기가 작고 많이 변하지 않으므로 이를 파일형태로 데이터웨어하우스로 로드하는 방법

Seeds는 작은 파일 데이터를 지칭한다. (보통 csv)

 

대충 굉장히 작은 csv 파일을 seeds/에 하나 만들었다.

 

굉장히 작은 csv파일이다.

 

dbt seed 명령어를 통해 DB에 적재할 수 있다.

 

 

colab에서 Redshift에 연결해서 테이블을 확인해보았다.

 

정상적으로 적재되었다.

 

---

🟦 DBT Sources

Staging 테이블들 만들 때 입력 테이블들이 자주 바뀐다면? > 각 테이블에 별칭 제공하는 기능

처음 입력이 되는 ETL 테이블들 대상으로 별칭을 제공하고 최신 레코드 체크 기능도 제공한다.

 

1. 별칭 제공

 

models/sources.yml 파일을 다음과 같이 수정

 

version: 2

sources:
  - name: sunhokim
    schema: raw_data
    tables:
      - name: metadata
        identifier: user_metadata
      - name: event
        identifier: user_event
      - name: variant
        identifier: user_variant

 

스키마와 테이블에 별칭을 제공해주어, SQL문을 작성할 때 별칭으로 불러올 수 있게 된다.

들여쓰기에 주의해서 작성하자.

 

그리고 테이블을 사용한 SQL 쿼리에서 템플릿형태로 별칭을 불러오면 된다.

 

ex)

 

변경 전

WITH src_user_event AS (
    SELECT * FROM raw_data.user_event
)

 

변경 후

WITH src_user_metadata AS (
    SELECT * FROM {{ source("sunhokim", "event") }}
)

 

위처럼 코드를 변경해도 전과 같이 똑같이 동작한다.

 

 

2. 최신성 (Freshness) 체크

models/sources.yml 파일을 다음과 같이 수정한다.

 

version: 2

sources:
  - name: sunhokim
    schema: raw_data
    tables:
      - name: metadata
        identifier: user_metadata
      - name: event
        identifier: user_event
        loaded_at_field: datestamp
        freshness:
          warn_after: { count: 1, period: hour } 
          error_after: { count: 24, period: hour }

      - name: variant
        identifier: user_variant

 

event 테이블의 datestamp 필드를 비교하여 데이터가 최신화된 데이터인지 확인할 수 있게된다.

 

freshness: 의 warn_after와 error_after:에서 datestamp가 얼마나 현재 시간과 멀어졌을때 경고나 오류를 보낼지 결정한다.

 

위에서 작성한 warn_after의 경우, 데이터의 datestamp가 현재 시간에서 1시간 ~ 24시간차이 날때 Warn를 주고,

error_after는 데이터의 datastamp가 현재 시간에서 24시간이상 차이날때 Error를 보낸다.

 

 

이는 dbt source freshness 명령어로 체크할수 있다.

 

현재 데이터에는 datestamp가 2023-06-22 10:00:00인 데이터가 하나 삽입되었고,

dbt source freshness를 명령한 시간은 2023-06-23 21:00:00이었다.

 

Error가 나타난다.

 

데이터의 datestamp가 명령을 실행한 시간보다 24시간이상 경과되었으니 Error가 나타나는 것이 맞다.

 

---

 

🟩 Dbt Snapshots

 

테이블의 변화를 계속적으로 기록함으로써 과거 시점의 테이블 내용을 확인할 수 있음

 

스냅샷을 저장하고 싶은 테이블은 snapshot/ 에 SQL문을 작성해서 저장하면 된다.

 

snapshot/scd_user_metadata.sql

{% snapshot scd_user_metadata %}
{{
    config(
    target_schema='sunhokim_public',
    unique_key='user_id',
    strategy='timestamp',
    updated_at='updated_at',
    invalidate_hard_deletes=True
    )
}}
SELECT * FROM {{ source('sunhokim', 'metadata') }}
{% endsnapshot %}

 

• target_schema : 스냅샷을 저장할 스키마
• unique_key : 스냅샷 테이블의 PK 지정
• strategy : 히스토리를 잡는 기준
• updated_at : 어느 컬럼을 기준으로 히스토리를 잡을지 설정
• invalidate_hard_deletes : True일 경우 원본 데이터가 삭제되어도 스냅샷에는 해당 데이터를 유지하게 됨

 

그리고 dbt snapshot 명령어로 스냅샷을 지정한 스키마에 테이블 이름으로 저장한다.

 

성공적으로 snapshot이 저장되었다.

 

colab에서 scd_user_metadata 내용을 확인해보면, 여러 필드가 더 생성된 것을 확인할 수 있다.

 

dbt_* 형태로 4개의 필드가 더 생성되었다.

 

dbt_vallid_from과 dbt_valid_to 필드를 통해 해당 레코드가 어느 기간동안 유효했는지 알 수 있도록 저장된다.

dbt_valid_from이 데이터가 저장된 시작날짜이고, dbt_valid_to가 저장된 마지막 날짜이므로,

가장 최신의 데이터일 경우 dbt_valid_to 필드의 값은 NULL로 저장된다.

 

이 상태에서 아까 스냅샷을 저장한 테이블의 데이터를 바꿔주었다.

 

UPDATE raw_data.user_metadata
SET age = '20-29', updated_at = GETDATE()
WHERE user_id = 99;

 

그리고 dbt snapshot 명령어를 실행하면, 업데이트 데이터에 따라 dbt_vaild_to가 업데이트될 것이다.

 

---

 

🟨 DBT TEST

 

데이터 품질을 테스트하는 방법, Generic과 Singular 테스트 방법으로 나뉜다.

 

1. 내장 일반 테스트 (Generic)

unique, not_null, accepted_values, relationships 등의 테스트 지원

models/schema.yml 파일 생성을 통해 구현

 

ex) dim_user_metadata 테이블의 user_id 컬럼에 중복성과 not null 체크하기

version: 2
models:
  - name: dim_user_metadata
    columns:
    - name: user_id
      tests:
        - unique
        - not_null

 

dbt test 명령어를 통해 간단하게 test가 가능하다.

 

 

2. 커스텀 테스트 (Singular)

SELECT으로 테스트를 짜서 결과가 리턴되면 실패로 간주하는 방법이다.

WHERE절에 해당하는 데이터가 있으면 데이터 품질에 문제가 있는 것으로 생각한다.

tests/ 폴더에 sql 파일을 생성함으로써 구현한다.

 

ex) tests/dim_user_metadata.sql

SELECT
 *
FROM (
    SELECT
    user_id, COUNT(1) cnt
    FROM
    {{ ref("dim_user_metadata") }}
    GROUP BY 1
    ORDER BY 2 DESC
    LIMIT 1
)
WHERE cnt > 1

이처럼 dim_user_metadata 테이블의 user_id 컬럼이 primary key uniqueness를 보장하도록 설계할 수 있다.

 

dbt test --select dim_user_metadata 형태로 특정 테이블만 test를 실행할 수도 있다.

 

성공적으로 테스트가 완료되었다.