📒 DBT(2)

📌DBT Seeds

• 많은 dimension 테이블들은 크기가 작고 많이 변하지 않는다.
• Seeds는 이를 파일 형태로 데이터웨어하우스로 로드하는 방법이다.
  -> Seeds는 작은 파일 데이터를지칭한다 (보통 csv파일)
• 'dbt seed'명령어를 입력해서 빌드한다.

---

📌DBT Sources

1. Staging 테이블을 만들 때 입력 테이블들이 자주 바뀐다면?

• models 아래 .sql 파일들을 일일이 찾아 바꾸어주어야 한다.
• 이 번거로움을 해결하기 위한 것이 Sources이다.
  -> 입력 테이블에 별칭을 주고 별칭을 staging 테이블에서 사용하는 것이다.

2. Sources 소개

• 기본적으로 처음 입력 되는 ETL 테이블을 대상으로 한다.

  • 별칭 제공
  • 최신 레코드 체크 기능 제공

• 테이블 이름들에 별명(alias)을 주는 것이다.

  • 이를 통해 ETL단의 소스 테이블이 바뀌어도 뒤에 영향을 주지 않는다.
  • 추상화를 통한 변경처리를 용이하게 하는 것이다.
  • 이 별명은 source 이름과 새 테이블 이름의 두 가지로 구성된다.
    -> ex) raw_data.user_metadata -> urface, metadata

• Source 테이블들에 새 레코드가 있는지 체크해주는 기능도 제공한다.

3. 실습

• models/sources.yml 생성

version: 2

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

• raw_data.user_metadata는 Jinja에서 source("urface", "metadata")로 지칭된다.

• 아래는 src_user_event.sql의 예시이다.
  -> models 아래의 다른 파일들도 적절하게 변경해야한다.

WITH src_user_event AS(
	SELECT * FROM {{ source("urface", "event") }}
)
SELECT
	 user_id,
     ....
     
FROM
	src_user_event

4. Sources 최신성(Freshness)

• 특정 데이터가 소스와 비교해서 얼마나 최신성이 떨어지는지 체크하는 기능
• dbt source freshness 명령어로 실행
• 이를 하려면 mldels/sources.yml의 해당 테이블 밑에 아래를 추가해야한다.

tables:
	- name: event
    	identifier: user_event
        loaded_at_field: datastamp
        freshness:
        	warn_after: { count: 1, period: hour }
            error_after: { count: 24, period: hour }

-> 지금 raw_data.user_event 테이블에서 datestamp의 최대값이 현재 시간보다 1시간 이상 뒤쳐저 있지만 24시간이 아니라면 Warning, 24시간 이상이라면 ERROR

---

📌 DBT Snapshots

• Dimension 테이블은 성격에 따라 변경이 자주 생길 수 있다
• bt에서는 테이블의 변화를 계속적으로 기록함으로써 과거 어느시점이건 다시 돌아가서 테이블의 내용을 볼 수 있는 기능을 말한다.
  • 이를 통해 테이블에 문제가 있을 경우 과거 데이터로 롤백 가능하다.
  • 다양한 데이터 관련 문제 디버깅도 쉬워진다.

1. SCD Type 2와 DBT

• Dimension 테이블에서 특정 entity에 대한 데이터가 변경되는 경우
  -> ex) employee_jobs 테이블
  • 특정 employee_id의 job_code가 바뀌는 경우
  • 변경시간도 같이 추가되어야 한다.
• Dimension 테이블에서 특정 entity에 대한 데이터가 변경되는 경우
• 새로운 Dimension 테이블을 생성(History/snapshot 테이블)

2. dbt의 스냅샷 처리 방법

❖ 먼저 snapshots 폴더에 환경설정이 됨
❖ snapshots을 하려면 데이터 소스가 일정 조건을 만족해야함
● Primary key가 존재해야함
● 레코드의 변경시간을 나타내는 타임스탬프 필요 (updated_at, modified_at 등등)
❖ 변경 감지 기준
● Primary key 기준으로 변경시간이 현재 DW에 있는 시간보다 미래인 경우
❖ Snapshots 테이블에는 총 4개의 타임스탬프가 존재
● dbt_scd_id, dbt_updated_at
● valid_from, valid_to

3. dbt snapshot 적용해보기

• snapshots/scd_user_metadata.sql 편집

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

4. dbt snapshot 실행

• dbt snapshot 명령어 입력

5. 실습

• raw_data.user_metadata 업데이트 실습 개요

  • 소스 테이블의 기존 레코드 하나의 age 그룹을 변경

  • 그리고 dbt snapshot 실행

  • urface.scd_user_metadata에 레코드 추가되었는지 확인

• 소스 테이블의 기존 레코드 하나의 age 그룹을 변경

SELECT *
FROM raw_data.user_metadata
WHERE user_id = 99;

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

• dbt snapshot 실행

• urface.scd_user_metadata에 레코드 추가되었는지 확인

SELECT * FROM urface.scd_user_metadata WHERE user_id = 99;

---

📌 DBT Tests

• 데이터 품질을 테스트하는 방법이다.

  • 내장 일반 테스트 (“Generic”)

    • unique, not_null, accepted_values, relationships 등의 테스트 지원한다.
    • models 폴더

  • 커스텀 테스트 (“Singular”)

    • 기본적으로 SELECT로 간단하며 결과가 리턴되면 “실패”로 간주한다.
    • tests 폴더

📌 DBT Documentation

• 기본 철학은 문서와 소스 코드를 최대한 가깝게 배치하자는 것이다.

• 문서화 자체는 두 가지 방법이 존재한다.

  • 기존 .yml 파일에 문서화 추가 (선호되는 방식)
  • 독립적인 markdown 파일 생성

• 이를 경량 웹서버로 서빙한다.

  • overview.md가 기본 홈페이지가 된다.
  • 이미지등의 asset 추가도 가능하다.

📌 DBT Expectations

• Great Expectations에서 영감을 받아 dbt용으로 만든 dbt 확장판
  • https://github.com/calogica/dbt-expectations
• 설치 후 packages.yml에 등록
  packages:
• package: calogica/dbt_expectations
  version: [">=0.7.0", "<0.8.0"]
• 보통은 앞서 dbt 제공 테스트들과 같이 사용
  • models/schema.yml