[TIL] 데이터분석 데브코스 69일차 (1) - Database Normalization/DBT(Data Build Tool)/ELT/DBT Seeds/DBT Sources, freshness/DBT Snapshots/DBT Tests/DBT Documentations/DBT Expectations

Database Normalization (데이터베이스 정규화)

: 데이터베이스 정합성을 쉽게 유지하고 레코드의 수정/적재/삭제를 용이하게 하는 과정

→ 데이터베이스를 보다 조직적이고 일관되게 설계 가능

 

용어

• Primary Key
• Composite Key
• Foreign Key : 어떤 테이블에서 다른 테이블의 pk를 사용하려 할 때 참조하는 키

 

제 1 정규화 (1NF; First Normal Form)

• Atomicity : 하나의 셀에는 하나의 값만 있어야 함 
• Primary Key가 있어야 함
• 중복된 키 or 레코드가 없어야 함

 

제 2 정규화 (2NF; Second Normal Form)

• 1NF를 만족하면서
• Primary Key를 중심으로 의존 결과를 알 수 있어야 함
• 부분적인 의존도가 없어야함
• 모든 부가 속성들은 Primary key를 갖고 찾을 수 있어야 함

 

제 3 정규화 (3NF; Third Normal Form)

• 2NF를 만족하면서
• 전이적 부분 종속성이 없어야 함

 

 

 

 

 

 

SCD (Slowly Changing Dimensions)

: DW나 DL에서는 모든 테이블들의 히스토리를 유지하는 것이 중요한데, 보통 2 개의 timestamp 필드를 갖는 것이 좋다.

• created_at : 생성 시간. 한 번 만들어지면 고정됨
• updated_at : 마지막 수정시간. 필수

 

 

SCD Type 0

• 히스토리와 상관 X
• 한 번 쓰고 나면 바꿀 이유가 없는 경우 (고정 필드)
• ex. 고객 테이블의 경우 회원 등록일, 제품 첫 구매일

 

SCD Type 1

• 데이터가 새로 생기면 덮어쓰면 되는 필드
• 처음 레코드 생성시에는 존재 X → 나중에 생기면서 채우는 경우
• ex. 고객 테이블의 경우 연간소득 필드

 

SCD Type 2

• 특정 entity에 대한 데이터가 새로운 레코드로 추가되는 경우
• 변경 시간도 같이 추가되어야 함
• ex. 고객 테이블의 경우 고객의 등급 변화

 

SCD Type 3

• SCD Type 2의 대안
• 특정 entity 데이터가 새로운 컬럼으로 추가되는 경우

 

SCD Type 4

• SCD Type 2의 다른 버전
• 특정 entity에 대한 데이터를 새로운 Dimension 테이블에 저장하는 경우
• 일반적으로 사용함
• ex. 별도 테이블로 저장하고 이걸 아예 일반화도 가능

 

 

 

 

 

 

 

 

 

 

DBT (Data Build Tool)

• ELT용 오픈소스 (In-warehouse Data Transformation)
• 다양한 데이터 웨어하우스 지원 : Redshift, Snowflake, Big Query, Spark
• 클라우드 버전 존재 : dbt Cloud

 

 

DBT 구성요소

1. 데이터 모델 (models)

• 테이블들을 몇 개의 tier로 관리 : Bronze, Staging, Core Table
• Table, View, CTE 등으로 존재함

2. 데이터 품질 검증 (tests)

 

3. 스냅샷 (snapshots)

 

 

 

 

 

DBT 사용 시나리오

DBT가 어떻게 사용될 수 있는지 가상 환경을 생각해보자.

• 데이터 변경 사항을 이해하기 쉽게 해야한다면 ROLLBACK 가능
• 데이터 간 리니지(Lineage) 확인 간으
• 데이터 품질 테스트 및 에러 보고
• Fact 테이블의 증분 로드 (Incremental Update)
• Dimension 테이블 변경 추적 (히스토리 테이블)
• 용이한 문서 작성

 

 

 

 

DBT에서 사용하는 테크 스택

• Data Warehouse : Redshift/Spark/Snowflake/BigQuery
• DBT
• 스케줄러 : Airflow (DBT를 주기적으로 실행시켜줌)

 

 

 

 

 

 

 

---

Fact 테이블

: 분석의 초점이 되는 양적 정보를 포함하는 중앙 테이블

• 일반적으로 매출액, 판매량, 이익 등이 포함 → 비즈니스 결정에 사용
• FK를 통해 여러 Dimension 테이블과 연결됨
• 변하지 않는 정보

ex. 위의 예제에서 user_event 테이블

 

 

Dimension 테이블

: Fact 테이블에 대한 상세 정보를 제공하는 테이블

• 일반적으로 고객, 제품 정보 등이 포함
• Fact 테이블의 데이터에 맥락을 제공 → 다양한 방식으로 분석 가능
• PK를 가지며 Fact 테이블에서 이를 참조
• 시간이 지나면서 변할 수 있는 정보
• 보통 Fact 테이블보다 크기가 훨씬 작음

ex. 위의 예제에서 user_metadata 테이블

 

 

 

 

 

 

 

 

 

 

ELT 실습

문제 정의

• 목표 : Redshift를 사용해서 AB 테스트 분석을 쉽게 하기 위한 ELT 테이블 생성하기
• 입력 테이블 : user_event, user_variant, user_metadata
• GROUP BY : variant_id, user_id, datestamp, age, gender
• 출력 테이블 : Variant별 사용자별 일별 요약 테이블 (총 impression, 총 click, 총 purchase, 총revenue를 요약)

 

 

테이블 생성

1) user_event

: 사용자/날짜/아이템별로 impression이 있는 경우 그 정보를 기록하고, impression으로부터 클릭/구매/구매 시 금액을 기록한 정보

CREATE TABLE raw_data.user_event (
	user_id int,
	datestamp timestamp,
    item_id int,
    clicked int,
    purchased int,
    paidamount int
);

 

 

2) user_variant

: 사용자가 소속된 AB Test Variant를 기록한 파일

CREATE TABLE raw_data.user_variant (
	user_id int,
	variant_id varchar(32) -- control vs. test
);

 

 

3) user_metadata

: 사용자별 메타 정보

CREATE TABLE raw_data.user_metadata (
	user_id int,
	age varchar(16),
	gender varchar(16),
	updated_at timestamp
);

 

 

 

 

 

최종 생성된 ELT 테이블

SELECT variant_id,
		ue.user_id,
		datestamp,
		age,
		gender,
		COUNT(DISTINCT item_id) num_of_items, -- 총 impression
		COUNT(DISTINCT CASE WHEN clicked THEN item_id END) num_of_clicks, -- 총 click SUM(purchased) num_of_purchases, -- 총 purchase
		SUM(paidamount) revenue -- 총 revenue
FROM raw_data.user_event ue
JOIN raw_data.user_variant uv ON ue.user_id = uv.user_id
JOIN raw_data.user_metadata um ON uv.user_id = um.user_id
GROUP by 1, 2, 3, 4, 5;

 

 

 

 

 

 

 

---

DBT 사용 프로세스

1. DBT 설치하기
   1. Cloud 버전 : dbt Cloud
   2. 로컬 개발 버전 : dbt Core
   3. git 사용
2. DBT 환경 설정
3. Connector(바탕이 되는 데이터 시스템) 설정 : Redshift, Spark 등
4. 데이터 모델링 : Raw Data → Staging Table → Core Table
5. 테스트 코드 작성
6. (필요시) Snapshot 설정

 

 

 

 

dbt init 프로젝트명

프로젝트는 ELT 작업 하나를 수행한다. (파일형식 : yml)

 

outputs : 프로젝트를 만들때 사용할 수 있는 다양한 환경 설정 가능

개발용 환경을 사용할것이라면 target : dev

 

 

 

 

 

 

---

DBT 설치하기 (로컬 개발 버전 : dbt Core)

 

 

터미널에 `pip3 install dbt-redshift`를 입력하자.

(에러가 발생한다면 `sudo pip3 install dbt-redshift`로 입력해보자.)

이 명령어는 dbt-core 모듈을 설치해줌과 동시에 환경에 맞는 dbt connector(Redshift, BigQuery, Snowflake 등)도 설치해준다.

 

본인 컴퓨터에 맞는 시스템 암호를 입력해준다.

 

 

 

설치가 완료되었다면 dbt의 버전을 확인해보자. 

`dbt --version`

 

 

최신 버전임을 확인했다.

 

 

 

이제 ELT 작업을 할 프로젝트를 만들어보자.

`dbt init 프로젝트명`

 

여기서 하나의 프로젝트는 하나의 ELT 작업을 뜻한다.

 

필자는 실습에서 Redshift를 사용할 것이기 때문에 1을 입력해주었다.

 

그럼 사용할 Redshift 연결 정보를 입력하도록 주어진다.

본인 계정에 맞게 host주소, port 번호, Redshift 계정 사용자 ID를 입력하자.

비밀번호를 입력할 것이라면 1을 선택한 다음 Redshift 계정 비밀번호를 입력한다.

이후 순서대로 DB명, 스키마명을 입력하고 1을 입력한다.

 

이렇게 되면 초기 환경설정이 끝난다.

 

 

 

 

환경설정이 끝났다면 프로젝트가 있는 디렉토리 안으로 이동해보자.

`cd 프로젝트명`

`ls -tl`

 

그럼 위와 같이 생성한 프로젝트 밑에 있는 다양한 폴더들이 있음을 확인할 수 있다.

 

 

 

 

 

dbt_project.yml

프로젝트 폴더를 살펴보자.

 

`~/.dbt/profiles.yml` 안에 존재해야 하는 정보들이다.

name: 'learn_dbt'
version: '1.0.0'
config-version: 2

profile: 'learn_dbt'

 

 

 

환경 정보가 들어가 있다. 이는 폴더 이름들과 일치해야 한다.

model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]

 

 

 

target, dpt_packages : 중간 결과들이 저장되는 폴더

target-path: "target" # folder to store compiled SQL files
clean-targets: # directories to be removed by `dbt clean`
	- "target"
    - "dbt_packages"
models:
	learn_dbt
    example:
    	+materialized: view

 

 

 

 

 

 

 

 

DBT Model

: ELT 테이블을 만듬에 있어 기본이 되는 빌딩 블록

• models라는 폴더에 들어가게 됨
• 물리적 테이블, 뷰, CTE(임시 테이블)의 형태로 존재
• 입력, 중간, 최종 테이블을 정의하는 곳
• 티어 : raw→ staging(src) → core

 

📌 View(뷰)
: SELECT 결과를 기반으로 만들어진 가상 테이블

- 기존 테이블의 일부 혹은 여러 테이블들을 JOIN한 결과 제공
- 문법 : CREATE VIEW 뷰이름 AS SELECT ...

[장점]
- 데이터 추상화 : 사용자는 뷰를 통해 필요 데이터에 직접 접근 (원본 데이터 알 필요 X)
- 데이터 보안 : 뷰를 통해 사용자에게 필요한 데이터만 제공 (원본 데이터 접근 필요 X)
- 쿼리 간소화 : SQL(View)를 사용하면 복잡한 쿼리를 단순화 가능

[단점]
- 물리적 테이블과 달리, 뷰의 기반이 되는 SELECT문이 매번 실행됨
- 원본 데이터의 변경을 모르면 실행 실패함

 

 

📌 CTE(Common Table Expression; 임시 테이블)

[문법]

WITH temp1 AS (
	SELECT k1, k2
    FROM t1
    JOIN t2 ON t1.id = t2.foreign_id
),
temp2 AS (
...
)
SELECT *
FROM temp1 t1 
JOIN temp2 t2 ON ...​

 

 

 

 

 

DBT Model 구성요소 : Input

: ELT에 입력되는 테이블로, 입력(raw)과 중간(staging, src) 데이터를 말함

 

• raw : CTE로 정의함
• staging : 뷰로 정의함
• 2번의 변환 과정을 거친다.
  • 1) data clean up
  • 2) data transform

 

DBT Model 구성요소 : Output

: ELT의 출력으로 나오는 테이블로, 최종(core) 데이터를 말함

• core : 테이블로 정의함

 

► Input과 Output은 모두 `models` 폴더 밑에 .sql 파일 형태로 존재한다.

• 기본적으로 `SELECT` + Jinja 템플릿(동적으로 다른 테이블에 접근 가능) + 매크로
• 다른 테이블 사용 가능 (references) → reference를 통해 리니지 파악

 

📌 Jinja 템플릿
: Python이 제공하는 템플릿 엔진

- Flask, Airflow 등에서 사용됨
- 입력 파라미터를 기준으로 HTML 페이지(마크업)을 동적으로 바꿔가면서 동일하게 사용할 수있도록 하는 기능 제공
- 조건문, 루프, 필터 등 제공

 

 

 

 

 

데이터 빌딩 프로세스

실습에서 만들어볼 테이블 구조

1. Raw Data

 

2. Staging

• 기본 품질을 체크한 후 staging으로 새로운 테이블을 생성
• `models/src/ ` 폴더 밑에 .sql 파일로 저장됨
• 이때 dbt 모델을 사용함
• 테이블, 뷰, CTE, Increment(Append) 형태로 존재할 수 있음

 

3. Core

• core 테이블을 JOIN 해서 최종 output으로 만듦
• `models/dim/ ` 폴더 밑에 .sql 파일로 저장됨

 

► 이 전체 과정을 Airflow DAG로 한번에 실행할 수 있다.

 

 

[Input]
모델 빌딩 명령어 : `dbt run`
기본적으로 뷰(View)의 형태로 만들어진다.
테스트를 실행하는 명령어 : `dbt test`
문서를 생성하는 명령어 : `dbt docs generate`

[Output]
core 테이블을 위한 폴더 생성 : `mkdir 폴더명`
해당 폴더로 이동 : `cd 폴더명`
폴더 밑에 테이블 생성 : `vi 테이블명.sql`

[Jinja 템플릿]
`{{ref('src_user_variant')}}` :  models 밑에 테이블명이 src_user_variant인 것을 생성하겠다는 의미

Incremental Table로 빌딩 (materialized = incremental)

{{
	config(
		materialized = 'incremental', -- incremental_strategy 사용 가능
		on_schema_change='fail' -- 스키마가 바뀐 경우 대응 방법 : 실패해라
	)
}}
WITH src_user_event AS (
	SELECT *
    FROM {{ ref("src_user_event") }}
)
SELECT
	user_id, datestamp, item_id, clicked, purchased, paidamount
FROM src_user_event​

[스키마가 바뀐 경우 대응 방법]
- `append_new_columns`
- `ignore`
- `sync_all_columns`
- `fail`

[materialized]
► incremental_strategy 사용가능
- `append`
- `merge`
- `insert_overwrite`

 

 

 

 

 

 

 

 

Materialization

: 입력 테이블들을 연결해서 새로운 테이블을 생성하는 것 = ELT

• 이 과정에서 data transformation, data clean-up 수행
• 파일 혹은 프로젝트 레벨에서 가능
• materialization 실행 명령어 : `dbt run`

 

Materialization 유형

1. View

• 데이터를 자주 사용하지 않는 경우

2. Table

• 데이터를 반복해서 자주 사용하는 경우

3. Incremental(Table Appends)

• Fact 테이블
• 과거 레코듣를 수정할 필요가 없는 경우

4. Ephemeral(CTE)

• 하나의 `SELECT`문에서 자주 사용되는 데이터를 모듈화하는데 사용

 

 

 

모델의 materialized format 결정

• 최종 core 테이블들은 뷰가 아닌 '테이블'로 빌드한다.
• `dbt_project.yml`을 편집

 

 

 

DBT compile vs. DBT run

• compile : 최종 .sql 형태로 결과를 만드는 것 (SQL 코드만 생성하고 실행은 X)
• run : 생성된 SQL 코드를 실제로 실햏ㅇ

 

 

 

CTE로 변환

src:
+materialized: ephemeral

 

 

 

 

 

---

DBT Seeds

: dimension 테이블을 csv  파일로 만들어서 데이터 웨어 하우스로 적재하는 기능

• dimension 테이블들은 크기가 작고 많이 변하지 않음
• 명령어 : `dbt seed`
• csv 파일은 프로젝트 생성시 만들었던 스키마 밑에 생성됨

 

실습

터미널을 실행해보자.

seeds 폴더 밑으로 들어가보자. `cd seeds`

 

 

seeds 파일명을 지정해서 생성해보자. `vi 시드파일명.csv`

 

 

데이터를 입력해보자.

date
2023-01-01
2023-01-02
2023-01-03
2023-01-04
2023-01-05
2023-01-06
2023-01-07
2023-01-08
2023-01-09
2023-01-10
2023-01-11
2023-01-12
2023-01-13

 

 

seeds 파일을 실행해보자. `dbt seed`

 

 

실제로 잘 생성되었는지 확인해보자. 를 켜서 아래의 명령어를 실행하면 된다.

SELECT *
FROM 스키마명.시드파일명;

 

 

 

 

 

 

 

DBT Sources

: 처음 입력이 되는 ETL 테이블을 대상으로 테이블 이름에 별칭(alias)를 부여하는 기능

 

Q. DBT Sources가 필요한 이유?

A. staging 테이블을 만들 때 입력 테이블이 자주 변경된다면?
→ models 폴더 밑에있는 .sql 파일들을 일일이 찾아서 바꿔줘야하며 이는 번거롭기도 하고 실수를 유발할 수 있다.
→ Sources를 통해 입력 테이블에 별칭을 부여하고, 해당 별칭을 staging 테이블에서 사용

 

 

• 별칭 제공
  • (2가지) : source 이름, 새 테이블 이름
  • 추상화를 통한 변경 처리를 용이하게 함
  • ETL 단의 source 테이블이 변경되어도 뒤에 영향주지 X
• 대상 테이블에 있는 최신 레코드 체크 기능

 

실습

`cd models`

`vi sources.yml`

identifier을 name으로 부르겠다는 의미

 

{{ source ("스키마명", "테이블명" }}

 

 

Sources 최신성 (Freshness)

: 특정 데이터가 source와 비교해서 얼마나 최신성이 떨어지는지 체크하는 기능

• 명령어 : `dbt source freshness`
• models/sources.yml의 해당 테이블 밑에 추가해야 함

 

 

실습

models 폴더로 이동해보자. `cd models`

 

sources 파일을 지정하자. `vi sources.yml`

 

아래의 내용을 입력하고 enter하자.

version: 2
sources:
	- name: keeyong
	schema: raw_data
    tables:
		- name: event
		identifier: user_event -- 지금 raw_data.user_event 테이블에서
        loaded_at_field: datestamp -- datestamp의 최대값이
        freshness:
			warn_after: { count: 1, period: hour } -- 현재시간보다 1시간이상 뒤쳐져있지만 24시간은 아니라면 warning
            error_after: { count: 24, period: hour } -- 24시간 이상이라면 error

 

src 폴더로 이동해보자.

`cd src`

`ls -tl`

 

세 개의 파일을 각각 변경해보자.

`vi src_user_metadata.sql`

SELECT *
FROM {{ source("스키마명", "metadata") }}
SELECT user_id,
		age,
        gender,
        updated_at
FROM src_user_metadata

 

`vi src_user_variant.sql`

SELECT *
FROM {{ source("스키마명", "variant") }}
SELECT user_id,
		variant_id
FROM src_user_variant

 

 

`vi src_user_event.sql`

SELECT *
FROM {{ source("스키마명", "event") }}
SELECT user_id,
		datestamp,
        item_id,
        clicked,
        purchased,
        paidamount
FROM src_user_event

 

 

이제 루트 디렉토리로 이동해보자. (dbt 명령어는 기본적으로 루트 디렉토리(프로젝트명)에서 실행해야 함)

`cd ..`

`cd..`

 

실행시켜보자.

`dbt run`

 

의미 : 3개의 source가 identified 되었다.

 

 

다시 models로 들어가보자.

`cd models`

 

소스 파일을 열어보자.

`vi sources.yml`

 

아래와 같이 세팅하고

version: 2
sources:
	- name: 스키마명
	schema: raw_data
    tables:
    	- name: metadata
        	identifier: user_metadata
		- name: event
			identifier: user_event -- 지금 raw_data.user_event 테이블에서
        	loaded_at_field: datestamp -- datestamp의 최대값이
        	freshness:
				warn_after: { count: 1, period: hour } -- 현재시간보다 1시간이상 뒤쳐져있지만 24시간은 아니라면 warning
            	error_after: { count: 24, period: hour } -- 24시간 이상이라면 error
        - name: variant
        	identifier: user_variant

 

 

최신성 기능을 실행해보자.

`dbt source freshness`

이렇게 입력하면 에러가 난다.

 

최신성 기능을 다시 실행해보자.

`dbt source freshness`

이 경우 에러는 나지 않지만 경고창이 뜬다.

 

 

 

 

 

 

 

 

 

DBT Snapshots

: Dimension 테이블은 특성상 변경이 자주 생길 수 있는데, 이러한 테이블의 변화를 지속적으로 기록해서 과거 어느 시점으로 닷 ㅣ돌아가 테이블의 내용을 언제든지 확인할 수 있는 기능

• 테이블에 문제가 있을 경우 과거 데이터로 rollback 가능
• 다양한 데이터 관련 문제를 디버깅할 수 있음

 

 

DBT Snapshots 처리 방법

1. snapshots 폴더 밑에 대상이 되는 .sql 파일을 생성

2. 

updated_at 보다 더 최신의 데이터가 있다면 

 

 

 

실습

snapshots 폴더로 이동해보자.

`cd snapshots`

`ls -tl`

 

scd_user_metadata.sql 파일을 만들어보자.

`vi scd_user_metadata.sql`

 

 

파일 내에 아래의 내용을 입력하자.

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

{% endsnapshot %}

 

 

루트 디렉토리로 이동하자.

`cd ..`

 

스냅샷 기능을 실행하보자.

`dbt snapshot`

 

포스티코에서 지금까지 수행한 것을 살펴보자.

 

 

포스티코 SQL 에디터로 들어가보자.

 

 

 

데이터를 변경해보자. 

 

 

SELECT문을 실행해서 실제로 데이터가 바뀌었는지 살펴보자.

 

 

 

다시 터미널로 돌아가서 스냅샷 기능을 실행하자

`dbt snapshot`

 

 

다시 포스티코로 돌아가서 히스토리 테이블에 어떻게 결과가 나오는지 살펴보자.

 

 

 

 

 

 

 

 

 

DBT Tests

데이터 품질을 테스트하는 방법에는 2가지가 있다.

1. 내장된 일반 테스트 Generic

• unique, not_null, accepted_values, relationships 등의 테스트를 지원한다.
• models 폴더 밑에 만들어진다. → `models/schema.yml` 파일 생성
• 테스트 실행 명령어 : `dbt test`

 

2. 커스텀 테스트 Singular

• 기본적으로 `SELECT`로 간단하며 결과가 리턴되면 "실패"로 간주한다.
• tests 폴더 밑에 만들어진다. → `tests/dim_user_metadata.sql` 파일 생성
• 테스트 실행 명령어 : `dbt test`

 

 

 

 

 

 

 

 

 

 

DBT Documentation

: 문서와 소스코드를 최대한 가깝게 배치하기 위한 기능

 

문서화 방법에는 2가지가 있다.

1. 기존 `.yml` 파일에 문서화 추가하는 방법

• `models/schema.yml` 이나 `models/sources.yml`에 description 키를 추가
• 문서 만들기 : `dbt doc generate` 명령어 수행
  • 사용자 권한이 있다면 Redshift로부터 더 많은 정보를 가져와 보여준다.
  • 결과 파일은 `target/catalog.json` 파일이 된다.
•  문서 보기 : `dbt docs serve` 명령어 수행 → 웹서버를 띄운다.
• 우측 하단에 청록색 버튼을 클릭하면 Lineage Graph를 확인할 수 있다.

 

2. 독립적인 markdown 파일을 생성하는 방법

 

이를 경량 웹서버로 서빙하는 명령어 : `dbt doc generate` 

• `overview.md`가 기본 홈페이지가 된다.
• 이미지 등의 asset을 추가할 수 있다.

 

 

 

 

 

DBT Expectations

: Great Expectations에서 영감을 받아 dbt용으로 만든 dbt 확장판 모듈

• 설치 후 `packages.yml`에 등록한다.
• 보통은 앞서 dbt 제공 테스트들과 같이 사용한다.