DBT를 도입한대요 (2) - 기초편

지난 글은 DBT의 개념에 대해서 담았다.

https://raiknow.tistory.com/128

DBT를 도입한대요 (1)

 

이번 글은 DBT의 주요 사용방법과 Airflow와 연동하는 방법에 대해 담을 예정이다.

---

DBT 주요 기능 사용법

dbt init을 하고나면 아래와 같은 디렉터리 구조가 생긴다. 

├── README.md
├── analysis
├── data
├── dbt_project.yml
├── macros
├── models
│   └── example
│       ├── my_first_dbt_model.sql
│       ├── my_second_dbt_model.sql
│       └── schema.yml
├── snapshots
└── tests

 

주요 파일 및 디렉터리는 아래와 같다.

1. dbt_project.yml
   • 이 yaml 파일이 있어야 현 디렉터리를 dbt 프로젝트로 인지한다.
   • 연결할 DB마다 형식이 다르므로 참고해서 작성해야한다.
   • profile.yaml 내의 어떤 프로필을 사용할 지 선택하는 곳도 이 파일이다.
   • 모델을 통해 생성되는 것이 테이블인지 View인지 Materialized  View인지 디폴트값을 설정해두는 곳도 이곳이다.
2. models/*
   
   • models 디렉터리 아래에 sql문(혹은 python 파일) 저장한다.
   • data transform을 위한 쿼리 및 코드를 모아두는 곳으로 파일 하나 당 모델 하나로 간주한다. 모델명은 파일명과 동일하므로 같은 제목의 파일이 존재하면 에러가 난다. 즉, 모든 파일 이름은 unique 해야함!
   • 모델 하나당 테이블(뷰)가 하나가 생성된다.
   • 기본적으로 형식은 dbt_proejct.yml 내의 디폴트 값을 따라가나 만일 특정 쿼리만 다른 형태로 저장하고싶다면 위에 config를 달아서 테이블(뷰) 인지 명시해주면 된다.
3. models/source.yml
   • source가 되는 모든 테이블 모아두고 추후에 jinja로 불러올 때 사용할 수 있다.
   • yaml 예시 : 
   • 사용 예시 : {{ source('source_name', 'table1') }} 
4. models/schema.yml
   • model(최종 table(뷰) 형태) 의 스키마를 저장하여 관리한다.
   • test를 추가하여 column별로 unique, not null등 체크가 가능하다.
   • yaml 예시 : 
   • tests key가 존재하면 dbt test를 실행시킬 때 자동으로 인식하여 지정한 column이 특정 조건을 만족하는지 검증해준다.
5. macros/*
   • 반복되는 문구는 이 곳에 저장해두고 재사용 할 수 있다.
   • 매크로 파일 예시 :  

     

   • 사용 예시 : {{ get_max_value(source('source_name', 'table1'), 'column_name') }}

이쯤하면 DBT를 가볍게 사용기엔 충분하다!

 

---

DBT와 Airflow 연동

DBT와 airflow를 연동하는 방법은 글을 따로 파야할 정도로 다룰 내용이 많지만, 이 글은 언제까지나 완전 핵심만 전달하는 글이기 때문에 간략하게 설명할 예정이다.

 

사용할 패키지는 astronomer-cosmos 이다. 3개 정도의 연동 툴( airflow-dbt , airflow-dbt-python, astronomer-cosmos )이 있길래 살펴봤는데 기능면에선 cosmos가 가장 뛰어나서 이 패키지로 선택했다.

 

1. dag/dbt/ 디렉터리 하위에 dbt project 구성

• airflow가 dbt project를 인지시키기 위한경로이다. dag/dbt/ 아래에 모든 dbt 프로젝트가 속해있어야한다. 
  (방법1) 터미널로 해당 경로에서 dbt init 명령어 실행
  (방법2) dbt init 시 생성되는 폴더 구조를 직접 생성, dbt_project.yml 파일이 존재하면 됨
  
   

2. dag 파일 생성

• DbtTaskGroup 클래스 내부에 아래 코드와 같이 프로젝트명 명시하면 자동으로 Task Group이 생성된다.(진짜 짱이다)

def dbt_test() -> None:
    """
    The simplest example of using Cosmos to render a dbt project as a TaskGroup.
    """
    dbt = DbtTaskGroup(
        group_id="dbt",
        project_config=ProjectConfig(
            (DBT_ROOT_PATH / "dbt_project_name").as_posix(),
        ),
        profile_config=profile_config,
    )
 
    dbt
 
dbt_test()

 

 

2.1. models 폴더 하위의 모델 파일에 따라서 자동으로 Run Task가 생성된다.

• 아래 예시 사진은 models 폴더 하위에 sql 파일이 2개있을 때 두개의 Run Task 생성된 모습이다.
• 의존성에 따라서 직렬/병렬도 자동으로 처리해주는데 기본적으론 병렬로 모든 작업이 생성된다. 
• model 간에 의존성이 있는 경우 직렬로 연결되는데 의존성을 주는 방법은 다음과 같다.
  • ref 사전 정의 매크로를 사용해서 의존성을 가질 모델 혹은 테이블(뷰)를 언급해준다.
  • depends_on: {{ ref('model_file_name') }} 와 같은 문구를 맨 마지막에 추가하면 언급 모델과의 의존성을 생성한다.
  • 혹은 depens_on 이라는 문구 떼고 앞서 생성한 테이블(뷰) 자체를 ref를 통해서 언급하면 자동으로 연결된다!

2.2 schema.yml에 Test에 관한 내용도 있다면 DBT Test를 수행하는 Task도 Run Task 다음에 자동 생성된다.

• 아래 예시는 sql파일 하나와 schema.yml에 Test 항목을 추가했을 경우 구성된 예시이다.
• 별도의 처리 없이 cosmos가 자동으로 Test Task를 Run Task 뒤에 연결한다.

 

(참고) models 하위에 디렉터리를 여러개 파서 관리하고 있다면 특정 디렉터리만 DAG로 구성되도록 설정할 수 있다. DAG를 정의할 때 아래와 같은 설정을 추가하면 원하는 디렉터리만 선택할 수 있다.

  render_config=RenderConfig(
        select=["path:models/~~~"],
    )

 

bashOperator등을 통해 직접 DBT 명령어로 Task 를 생성할 수도 있겠지만 cosmos가 자동으로 그래프를 그려주니 편리함이 아주 높았다. 이 정도가 Airflow와 DBT를 연결할 때 알아야 할 기본 정보 되시겠다 ^__^

 

---

DBT는 확실히 러닝커브는 높지만 한번 익히고 나면 여러 데이터 변환 과정을 한눈에 관리할 수 있다는 장점이 확실해보인다. 다음 글은 기회가 된다면 심화편으로 작성해보려고 한다. 끝!