Apache Spark configurations

dbt-spark 플러그인에서 모델을 구성하는 방법을 다루는 페이지예요. 특히 incremental 전략(append, insert_overwrite, merge, microbatch)과 파일 형식 설정에 초점을 맞춰요.

출처: 문서

본문

Databricks를 쓴다면 dbt-spark보다 dbt-databricks 어댑터를 권장해요. Databricks에서 여전히 dbt-spark를 쓰고 있다면 dbt-spark 어댑터에서 dbt-databricks 어댑터로의 마이그레이션을 고려하세요. 이 페이지의 Databricks 버전은 Databricks setup을 참고하세요.

테이블 구성

모델을 table로 materialize할 때, 표준 모델 config 외에도 dbt-spark 플러그인 고유의 선택적 config를 여러 개 포함할 수 있어요.

Incremental models

dbt는 내장 config와 materialization으로 유용하고 직관적인 모델링 추상화를 제공하려 해요. 세상의 Apache Spark 클러스터는 저마다 차이가 크고, Delta 파일 형식과 커스텀 런타임이 Databricks 사용자에게 제공하는 강력한 기능까지 고려하면 모든 옵션을 이해하는 것만으로도 큰 작업이에요.

또는 Apache Spark 런타임에서 Apache Iceberg나 Apache Hudi 파일 형식으로 incremental 모델을 만들 수도 있어요.

그래서 dbt-spark 플러그인은 incremental_strategy config에 크게 의존해요. 이 config는 incremental materialization이 첫 실행 이후의 실행에서 어떻게 모델을 빌드할지 알려줘요. 다음 세 가지 값 중 하나로 설정할 수 있어요.

append (기본값): 기존 데이터를 갱신하거나 덮어쓰지 않고 새 레코드만 삽입.
insert_overwrite: partition_by가 지정되면 테이블의 그 파티션을 새 데이터로 덮어써요. partition_by가 지정되지 않으면 전체 테이블을 새 데이터로 덮어써요.
merge (Delta, Iceberg, Hudi 파일 형식만): unique_key를 기준으로 레코드를 매칭하고, 기존 레코드는 갱신하고 새 것도 삽입해요. (unique_key를 지정하지 않으면 append처럼 모든 새 데이터가 삽입돼요.)
microbatch: event_time로 시간 기반 범위를 정의해 데이터를 필터링하는 마이크로배치 전략을 구현해요.

각 전략에는 장단점이 있고, 아래에서 다룰게요. 다른 모델 config와 마찬가지로 incremental_strategy는 dbt_project.yml이나 모델 파일의 config() 블록에서 지정할 수 있어요.

`append` 전략

append 전략에 따라 dbt는 모든 새 데이터로 insert into 문을 실행해요. 이 전략의 매력은 모든 플랫폼, 파일 타입, 연결 방식, Apache Spark 버전에서 단순하고 동작한다는 점이에요. 다만 기존 데이터를 갱신·덮어쓰기·삭제할 수 없어서, 많은 데이터 소스에서 중복 레코드를 삽입할 가능성이 있어요.

append를 incremental 전략으로 지정하는 건 선택 사항이에요. 지정하지 않을 때 사용되는 기본 전략이기 때문이에요.

spark_incremental.sql

{{ config(
    materialized='incremental',
    incremental_strategy='append',
) }}

--  All rows returned by this query will be appended to the existing table

select * from {{ ref('events') }}
{% if is_incremental() %}
  where event_ts > (select max(event_ts) from {{ this }})
{% endif %}

spark_incremental.sql (컴파일 결과)

create temporary view spark_incremental__dbt_tmp as

    select * from analytics.events

    where event_ts >= (select max(event_ts) from {{ this }})

;

insert into table analytics.spark_incremental
    select `date_day`, `users` from spark_incremental__dbt_tmp

`insert_overwrite` 전략

이 전략은 모델 config의 partition_by 절과 함께 지정할 때 가장 효과적이에요. dbt는 쿼리에 포함된 모든 파티션을 동적으로 교체하는 원자적 insert overwrite 문을 실행해요. 이 incremental 전략을 쓸 때는 파티션의 모든 관련 데이터를 다시 선택해야 합니다.

partition_by가 지정되지 않으면 insert_overwrite 전략은 테이블 전체 내용을 원자적으로 교체하고, 기존 데이터를 새 레코드로만 덮어써요. 다만 테이블의 컬럼 스키마는 그대로 유지돼요. 테이블 내용이 덮어써지는 동안 다운타임을 최소화하므로 제한된 상황에서는 유용할 수 있어요. 다른 데이터베이스에서 truncate + insert를 실행하는 것과 비슷한 연산이에요. Delta 형식 테이블의 원자적 교체는 table materialization(create or replace 실행)을 대신 사용하세요.

사용 참고:

이 전략은 file_format: delta 테이블에서는 지원되지 않아요.
Databricks SQL 엔드포인트(method: odbc + endpoint)로 연결할 때는 사용할 수 없어요.
Databricks 클러스터 + ODBC 드라이버(method: odbc + cluster)로 연결한다면, 동적 파티션 교체(incremental_strategy: insert_overwrite + partition_by)가 동작하려면 클러스터 Spark Config에 set spark.sql.sources.partitionOverwriteMode DYNAMIC을 반드시 포함해야 해요.

spark_incremental.sql

{{ config(
    materialized='incremental',
    partition_by=['date_day'],
    file_format='parquet',
    incremental_strategy='insert_overwrite'
) }}

/*
  Every partition returned by this query will be overwritten
  when this model runs
*/

with new_events as (

    select * from {{ ref('events') }}

    {% if is_incremental() %}
    where date_day >= date_add(current_date, -1)
    {% endif %}

)

select
    date_day,
    count(*) as users

from events
group by 1

spark_incremental.sql (컴파일 결과)

create temporary view spark_incremental__dbt_tmp as

    with new_events as (

        select * from analytics.events

        where date_day >= date_add(current_date, -1)

    )

    select
        date_day,
        count(*) as users

    from events
    group by 1

;

insert overwrite table analytics.spark_incremental
    partition (date_day)
    select `date_day`, `users` from spark_incremental__dbt_tmp

`merge` 전략

사용 참고: merge incremental 전략은 다음을 요구해요.

file_format: delta, iceberg or hudi
delta 파일 형식은 Databricks Runtime 5.1 이상
Iceberg 또는 Hudi 파일 형식은 Apache Spark

dbt는 Snowflake와 BigQuery의 기본 merge 동작과 거의 동일해 보이는 원자적 merge 문을 실행해요. unique_key가 지정되면(권장) dbt는 키 컬럼에서 매칭되는 새 레코드 값으로 기존 레코드를 갱신해요. unique_key가 지정되지 않으면 dbt는 매칭 기준을 생략하고 모든 새 레코드를 삽입해요(append 전략과 비슷).

merge_incremental.sql

{{ config(
    materialized='incremental',
    file_format='delta', # or 'iceberg' or 'hudi'
    unique_key='user_id',
    incremental_strategy='merge'
) }}

with new_events as (

    select * from {{ ref('events') }}

    {% if is_incremental() %}
    where date_day >= date_add(current_date, -1)
    {% endif %}

)

select
    user_id,
    max(date_day) as last_seen

from events
group by 1

target/run/merge_incremental.sql (컴파일 결과)

create temporary view merge_incremental__dbt_tmp as

    with new_events as (

        select * from analytics.events

        where date_day >= date_add(current_date, -1)

    )

    select
        user_id,
        max(date_day) as last_seen

    from events
    group by 1

;

merge into analytics.merge_incremental as DBT_INTERNAL_DEST
    using merge_incremental__dbt_tmp as DBT_INTERNAL_SOURCE
    on DBT_INTERNAL_SOURCE.user_id = DBT_INTERNAL_DEST.user_id
    when matched then update set *
    when not matched then insert *

모델 설명 영속화

관계(relation) 레벨 문서 영속화는 dbt에서 지원돼요. docs 영속화 설정에 대한 자세한 내용은 docs를 참고하세요.

persist_docs 옵션이 적절히 설정되면 describe [table] extended 또는 show table extended in [database] like '*'의 Comment 필드에서 모델 설명을 볼 수 있어요.

항상 `schema`, 절대 `database` 아님

Apache Spark는 "schema"와 "database"라는 용어를 같은 뜻으로 사용해요. dbt는 database가 schema보다 더 높은 수준에 존재한다고 이해해요. 그래서 dbt-spark를 실행할 때 node config나 target 프로필에서 database를 사용하거나 설정하면 안 돼요.

dbt가 모델을 materialize할 스키마/데이터베이스를 제어하려면 schema config와 generate_schema_name 매크로만 사용하세요.

기본 파일 형식 구성

스냅샷과 merge incremental 전략 같은 고급 incremental 전략 기능을 이용하려면 모델을 테이블로 materialize할 때 Delta, Iceberg, Hudi 파일 형식을 기본 파일 형식으로 쓰고 싶을 거예요.

프로젝트 파일에서 최상위 구성을 설정하면 꽤 편리해요.

dbt_project.yml

models:
  +file_format: delta # or iceberg or hudi

seeds:
  +file_format: delta # or iceberg or hudi

snapshots:
  +file_format: delta # or iceberg or hudi

(dbt v1.11 이상 적용)

PyHive 연결의 재시도 처리

HTTP 또는 Thrift 연결 방식을 쓸 때, 오래 실행되는 쿼리에 대해 dbt가 폴링·타임아웃·연결 재시도를 처리하는 방식을 구성할 수 있어요. 이 설정은 쿼리가 무한정 매달리는 것을 막고, 쿼리 실행 중 연결 중단으로부터 자동 복구되도록 도와줘요. 세 가지 프로필 구성이 있어요. (상세 표는 원문의 프로필 구성 표를 참고하세요.)

어댑터는 특정 연결 예외(예: ConnectionResetError, BrokenPipeError, TTransportException)를 잡아내고, 연결 손실 시 새 커서로 재시도해요. 모든 재시도를 소진하면 dbt는 DbtRuntimeError를 발생시키고 프로필에서 query_retries를 늘리라고 제안해요.

각주

location_root를 구성하면 dbt는 create table 문에 location 경로를 지정해요. 이렇게 하면 Spark/Databricks에서 테이블이 "managed"에서 "external"로 바뀌어요. ↩

더 알아보기 (Learn more)

Databricks configuration — Databricks 버전 설정.
Incremental strategies — incremental 전략 개요.

Apache Spark configurations