dbtアダプター
dbtは、ELT（抽出、ロード、変換）ワークフローの変換段階を管理します。dbt-for-apache-dorisアダプターはdbtモデルをDoris SQLにコンパイルし、Doris Frontend（FE）のMySQLクエリポートを通じて、テーブル作成、変換、テスト、ドキュメント生成を実行します。
ソースデータは、Internal Catalogのテーブルまたは設定済みのExternal Catalogを通じて、Dorisから照会できる状態である必要があります。アダプターはデータを変換しますが、ソースデータの取り込みや同期は行いません。
dbt-for-apache-dorisはVeloDBが提供・保守しています。Apache Dorisプロジェクトの一部ではなく、Apache Dorisコミュニティがリリースまたは推奨するものでもありません。本番環境で使用する前にアダプターを評価し、パッケージとリリース情報の完全性を確認して、サードパーティープロジェクトのライセンスに従ってください。アダプターの問題はdbt-for-apache-dorisプロジェクトに報告してください。
環境要件
| コンポーネント | 要件 |
|---|---|
| Python | 3.10以降 |
| dbt Core | 1.12.x（アダプターとともにインストール） |
リリース済みバージョン
| バージョン |
|---|
| 1.1.1 |
| 1.1.0 |
インストール
以下のコマンドでは、公開されているv1.1.1リリースをインストール例として使用します。別のリリースをインストールする場合は、pipの要件指定に含まれるバージョンを置き換えてください。
分離されたPython環境にPyPIからアダプターをインストールします。
python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install "dbt-for-apache-doris==1.1.1"
dbt --version
Windows PowerShellの場合:
py -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
python -m pip install "dbt-for-apache-doris==1.1.1"
dbt --version
この例では、Pluginsセクション（dbt --versionの出力）にdoris: 1.1.1が含まれるはずです。dbt CoreとMySQL Connector/Pythonは依存パッケージとしてインストールされます。
接続の設定
dbtプロジェクトを作成します。
dbt init doris_demo
cd doris_demo
~/.dbt/profiles.ymlにDorisの出力設定を追加します。認証情報はバージョン管理の対象外にしてください。この例では、環境変数からパスワードを読み取ります。
doris_demo:
target: dev
outputs:
dev:
type: doris
host: 127.0.0.1
port: 9030
username: dbt_user
password: "{{ env_var('DORIS_PASSWORD') }}"
schema: analytics
threads: 4
| 設定 | 推奨値 | 動作 |
|---|---|---|
type | dorisを指定 | Dorisアダプターを選択します |
host | Doris FEのアドレスを明示的に指定 | 省略すると基となる認証情報の設定では127.0.0.1が使用されます。dbt initの入力にはデフォルト値がありません |
port | 通常は9030 | 省略すると9030が使用されます。これはFEのクエリポートであり、HTTPポートではありません |
username | 専用のDorisユーザーを明示的に指定 | 省略すると基となる認証情報の設定ではrootが使用されます。dbt initの入力にはデフォルト値がありません |
password | 環境変数から読み取る | デフォルトは空文字列です |
schema | ターゲットのDorisデータベースを指定 | dbt initのプロンプトのデフォルトはdbt、基となる認証情報の設定のデフォルトはNoneです |
threads | FEとワークロードの処理能力に合わせて設定 | dbt initのプロンプトのデフォルトは1です |
database | Internal Catalogでは省略し、外部ターゲットでは既存のDoris Catalogを指定 | v1.1.1ではDoris Catalogに対応します |
v1.1.1では、dbtのdatabase.schema.identifierはDorisのcatalog.database.tableに対応します。Internal Catalogを使用する場合はdatabaseを省略し、schemaにDorisデータベースを指定します。外部ターゲットの場合は、databaseに既存のDoris Catalogを、schemaにそのCatalog内のデータベースを指定します。Catalogが対応していれば、アダプターは存在しないターゲットデータベースの作成を試みることができます。dbtの実行アカウントには、ソースの読み取りと、ターゲットオブジェクトの作成、変更、削除、書き込みの権限が必要です。External Catalogへの書き込み可否はDorisコネクターにも依存します。
dbt_project.ymlが同じプロファイルを参照していることを確認します。
name: doris_demo
version: "1.0.0"
config-version: 2
profile: doris_demo
model-paths: ["models"]
models:
doris_demo:
+materialized: view
macOSまたはLinuxでは、次のコマンドで接続を検証します。
export DORIS_PASSWORD='your_password'
dbt debug
Windows PowerShellの場合:
$env:DORIS_PASSWORD = 'your_password'
dbt debug
最初のモデルの構築
次のSQLは、この例で使用するソーステーブルを作成します。レプリカ数1が適しているのは、単一BEの開発用クラスターに限られます。
CREATE DATABASE IF NOT EXISTS raw;
CREATE TABLE IF NOT EXISTS raw.orders (
order_id BIGINT,
customer_id BIGINT,
order_time DATETIME,
amount DECIMAL(18, 2),
status VARCHAR(20),
updated_at DATETIME,
is_valid BOOLEAN
)
DUPLICATE KEY(order_id)
DISTRIBUTED BY HASH(order_id) BUCKETS 1
PROPERTIES ("replication_num" = "1");
INSERT INTO raw.orders VALUES
(1, 101, '2026-08-01 10:00:00', 20.00, 'PAID', '2026-08-01 10:05:00', true),
(2, 102, '2026-08-01 11:00:00', 35.00, 'PAID', '2026-08-01 11:05:00', true);
models/sources.ymlでソースを宣言します。
version: 2
sources:
- name: raw
schema: raw
tables:
- name: orders
models/fct_daily_sales.sqlを作成します。
{{
config(
materialized='table',
duplicate_key=['order_date'],
distributed_by=['order_date'],
buckets=1,
properties={'replication_num': '1'}
)
}}
select
cast(order_time as date) as order_date,
sum(amount) as sales_amount
from {{ source('raw', 'orders') }}
group by cast(order_time as date)
models/schema.ymlにデータテストを追加します。
version: 2
models:
- name: fct_daily_sales
description: Daily sales summary
columns:
- name: order_date
description: Order date
data_tests:
- not_null
- unique
- name: sales_amount
description: Daily sales amount
data_tests:
- not_null
モデルとテストを実行します。
dbt build --select fct_daily_sales
ビルドが成功すると、analytics.fct_daily_salesが作成されます。本番環境のレプリカ数とバケット数は、クラスターの構成に合わせて設定してください。
ソースが別のDorisデータベースにある場合は、schemaを直接指定することを推奨します。
sources:
- name: finance
schema: finance_raw
tables:
- name: payments
v1.1.1では、ソースのdatabaseはDoris Catalogの名前を表します。Internal Catalog内の別のデータベースから読み取るには、上記のようにschemaのみを設定します。v1.1.0でデータベースをまたぐソースにdatabase: finance_rawを使用していたプロジェクトでは、アップグレード前にその値をschemaへ移してください。
v1.1.1でのExternal Catalog
アダプターはExternal Catalog内のソースを読み取り、dbt docs generate用にそのテーブルと列を検出できます。先にDorisでCatalogを作成し、その名前をソースのdatabaseフィールドに指定します。
sources:
- name: lakehouse
database: hive_catalog
schema: ods
tables:
- name: orders
{{ source('lakehouse', 'orders') }}はhive_catalog.ods.ordersに解決されます。schemaフィールドは、そのCatalog内のデータベース名です。アダプターはCatalog名を含むリレーション名に対応していますが、Dorisで実行できるDDLと書き込み操作は各Catalogコネクターによって決まります。
Icebergのターゲットを作成するには、まず書き込み可能なIceberg Catalogを設定します。既存のプロファイル出力設定に次のフィールドを指定します。
database: iceberg_catalog
schema: analytics
models/iceberg_sample.sqlを作成します。
{{ config(materialized='table', engine='iceberg') }}
select cast(1 as int) as id
明示的にINTへキャストすることで、Icebergの列型として使用できないTINYINTとして推論されるのを防ぎます。
dbt run --select iceberg_sampleを実行します。iceberg_catalog.analytics.iceberg_sampleが作成されます。Catalogは存在している必要があり、そのデータベースは既存かdbtによって作成可能で、ターゲットは書き込み可能である必要があります。engine='iceberg'は明示的なエンジン句を出力します。engineを省略すると、DorisがターゲットCatalogから推論します。Catalogに基づく推論はDoris 4.1.3で検証されています。以前のDorisリリースでは、明示的なエンジン指定が必要な場合があります。
| Icebergのターゲット | v1.1.1での動作 |
|---|---|
table | Icebergテーブルを作成します。再実行時はバックアップと名前変更によって置き換えます |
incremental | 以下で説明するappend、merge、insert_overwrite、microbatchに対応します |
| SeedとSnapshot | 対応しています。Snapshotの公開とSeedの全件更新には、バックアップと名前変更を使用します。通常の再ロードでは、非パーティションテーブルにINSERT OVERWRITE、パーティションテーブルにネイティブMERGEを使用し、後者はV2以降が必要です。V1のパーティション化されたSeedにはdbt seed --full-refreshを使用してください |
viewとmaterialized_view | ターゲットはInternal Catalogに置く必要があります。Internal CatalogのビューとMVはExternal Catalogのソースを照会できます |
Icebergのmerge、microbatch、パーティション化されたSeedの再ロードには、ネイティブのIceberg MERGE INTOに対応するDorisリリースと、Iceberg V2以降のテーブルが必要です。これらの処理経路はDoris 4.1.3で検証されています。DorisのドキュメントではIceberg MERGEは4.1.0以降、実験的機能とされています。外部テーブルの置き換えはアトミックではありません。名前変更の間は一時的に元の名前でアクセスできなくなり、失敗すると旧テーブルがバックアップ名で残る場合があります。本番環境で使用する前に、分離されたテストプロジェクトでDorisのリリース、Catalogコネクター、復旧手順を検証してください。
アダプターは、Table、View、Incremental、MVのターゲットを書き込む前に、モデルまたは列のconstraints宣言を拒否します。モデルコントラクトでは引き続き列名と型を検査します。書き込み後のデータ検証にはdata_testsを使用してください。たとえばnot_nullを指定できます。これらはデータベースが強制する制約を作成しません。
実行可能な五つのDorisデモでは、モデル、テスト、データベースをまたぐソース、インクリメンタル処理、スナップショット、非同期マテリアライズドビューを扱います。デモのセットアップではデフォルトでdbt-for-apache-doris==1.1.1を使用します。
マテリアライゼーション
dbt-for-apache-dorisは、主に次のマテリアライゼーションに対応します。
| マテリアライゼーション | Dorisのオブジェクトまたは動作 | 用途 |
|---|---|---|
view | Internal CatalogのDorisビュー | 常に最新のソースデータを読み取る軽量な変換 |
table | 毎回全件再構築するInternal CatalogのDuplicate Keyテーブル、または書き込み可能なIcebergテーブル | 安定したクエリ性能を必要とする、結果の規模が限定された処理 |
incremental | 戦略に応じてデータを追加、アップサート、または上書き | 大規模テーブルのインクリメンタル処理 |
materialized_view | Internal CatalogのDoris非同期マテリアライズドビュー | Dorisが管理する更新と透過的なクエリ書き換え |
ephemeral | 下流モデルでCTEにコンパイル | Dorisオブジェクトを作成せずにSQLを再利用 |
SeedとSnapshotでもDorisテーブルを作成できます。現時点でアダプターには独自のpartitionマテリアライゼーションはありません。パーティションの置き換えには、インクリメンタルのinsert_overwriteを使用します。以下の例では、単一BEの開発用クラスターでも実行できるようreplication_num=1を使用しています。実際の本番環境では、適切なレプリカ数を使用してください。
ビュー
{{ config(materialized='view') }}
select order_id, customer_id, amount
from {{ source('raw', 'orders') }}
where status = 'PAID'
実行のたびにビュー定義が更新されます。ビューはデータを保存しません。クエリのコストはビューのSQLと下流のクエリによって決まります。
テーブル
{{
config(
materialized='table',
duplicate_key=['event_date', 'event_id'],
partition_by=['event_date'],
partition_type='RANGE',
partition_by_init=[
"PARTITION p_before_202608 VALUES LESS THAN ('2026-08-01')",
"PARTITION p202608 VALUES LESS THAN ('2026-09-01')",
"PARTITION pmax VALUES LESS THAN (MAXVALUE)"
],
distributed_by=['event_id'],
buckets=16,
properties={'replication_num': '1'}
)
}}
select
cast(order_time as date) as event_date,
order_id as event_id,
status as event_type
from {{ source('raw', 'orders') }}
Internal Catalogのターゲットでは、tableマテリアライゼーションはDuplicate Keyテーブルを作成し、以降の実行時にそのテーブルを全件置き換えます。Icebergのターゲットでは、DorisのOLAPキーやデフォルトの分散設定を使用しません。テーブルモデルでは、次のテーブル作成設定を指定できます。
| 設定 | 型とデフォルト | 説明 |
|---|---|---|
duplicate_key | 文字列またはリスト。省略可能 | Duplicate Keyの列 |
partition_by | 文字列またはリスト。省略可能 | パーティション列 |
partition_type | RANGEまたはLIST。デフォルトはRANGE | partition_byとともに使用するパーティション方式 |
partition_by_init | 文字列のリスト。省略可能 | テーブル作成時にpartition_byとともに使用するDorisのパーティション定義 |
distributed_by | 文字列またはリスト。省略可能 | ハッシュ分散の列 |
buckets | 正の整数。ハッシュ分散を出力する場合のデフォルトは10 | ハッシュバケット数。distributed_byとともに指定した場合のみ使用 |
replication_num | 正の整数または数値文字列。省略可能 | レプリカ数。トップレベルの設定はproperties内の同じキーより優先されます |
properties | 辞書。省略可能 | DorisのPROPERTIESとして出力するキーと値のペア |
Duplicate Key列、パーティション列、分散列はモデルの出力に含まれ、Dorisのテーブル作成ルールを満たす必要があります。アダプターはpartition_by_initとpropertiesをCREATE TABLE文に出力し、Dorisがパーティション定義、プロパティ名、値を検証します。duplicate_keyやdistributed_byなどのOLAP固有のオプションはIcebergのターゲットには適しません。通常のtableモデルでは、現時点でAggregate Keyや独立したUnique Keyの設定は公開されていません。Unique Keyによるアップサートには、インクリメンタルのmergeを使用します。
インクリメンタル
append、merge、insert_overwriteを使用するインクリメンタルモデルは、次の二つの要素で構成されます。
- モデルSQL（通常は
is_incremental()を使用）が、今回の実行で返すソース行を決定します。 incremental_strategyが、そのバッチをターゲットに書き込む方法を決定します。
Microbatchではこのフィルタリング方式を使用しません。dbt Coreがバッチのコンテキストと上流リソースに設定されたevent_timeを基に時間フィルターを挿入します。
| 戦略 | Dorisのターゲットと動作 | 主な要件 |
|---|---|---|
append | OLAPのDuplicate Keyテーブルまたは書き込み可能なIcebergテーブルにINSERT INTO | unique_keyを設定しないでください |
merge | OLAPのUnique Keyテーブルには全列のINSERT INTOによるアップサート、IcebergにはネイティブのMERGE INTO | unique_keyが必要です。IcebergではネイティブMERGEへの対応とV2以降のテーブルも必要です |
insert_overwrite | OLAPではテーブル全体、名前を指定したパーティション、または動的パーティションの上書き。Icebergではテーブル全体、動的または静的なパーティションの上書き | unique_keyは指定できません。Icebergの静的な対象範囲には列と値のマッピングを使用します |
microbatch | OLAPでは正確に対応するRANGEパーティションを上書きし、IcebergではネイティブMERGEで一つのUTCウィンドウを置き換えます | event_time、batch_size、beginを設定します。IcebergではネイティブMERGEへの対応とV2以降のテーブルが必要です |
戦略を設定しない場合、unique_keyがあればmergeが選ばれ、なければappendが使用されます。アダプターはon_schema_changeの四つのモード（ignore、fail、append_new_columns、sync_all_columns）すべてに対応します。
delete+insertとdelete_insertは拒否されます。DorisのUnique Keyによるアップサートにはmergeを使用してください。OLAPのターゲットでは、mergeはDorisのUnique Keyのセマンティクスに従い、全列のINSERT INTOを使用します。Icebergのターゲットでは、ネイティブのMERGE INTOを生成し、書き込み前にソースキーの重複を拒否します。各バッチで、unique_keyの各値に対応する行は最大一行でなければなりません。Sequence Columnを使用するには、function_column.sequence_colで可視列をproperties内に指定し、引き続きmergeを使用します。単独のsequence_col設定と、function_column.sequence_typeを隠し列__DORIS_SEQUENCE_COL__とともに使う方法には現時点で対応していません。
次の例では、order_idをキーとして全列をアップサートします。
{{
config(
materialized='incremental',
incremental_strategy='merge',
unique_key=['order_id'],
distributed_by=['order_id'],
buckets=16,
properties={'replication_num': '1'},
on_schema_change='append_new_columns'
)
}}
select
order_id,
customer_id,
amount,
updated_at
from {{ source('raw', 'orders') }}
{% if is_incremental() %}
where updated_at >= (
select coalesce(max(updated_at), '1970-01-01 00:00:00')
from {{ this }}
)
{% endif %}
名前を指定したInternal CatalogのDorisパーティションを上書きするには、次のようにします。
{{
config(
materialized='incremental',
incremental_strategy='insert_overwrite',
duplicate_key=['event_date', 'event_id'],
partition_by=['event_date'],
partition_type='RANGE',
partition_by_init=[
"PARTITION p_before_202607 VALUES LESS THAN ('2026-07-01')",
"PARTITION p202607 VALUES LESS THAN ('2026-08-01')",
"PARTITION p202608 VALUES LESS THAN ('2026-09-01')",
"PARTITION pmax VALUES LESS THAN (MAXVALUE)"
],
overwrite_partitions=['p202607', 'p202608'],
distributed_by=['event_id'],
buckets=16,
properties={'replication_num': '1'}
)
}}
select event_date, event_id, event_type
from (
select
cast(order_time as date) as event_date,
order_id as event_id,
status as event_type,
order_time
from {{ source('raw', 'orders') }}
) events
{% if is_incremental() %}
where order_time >= '2026-07-01'
and order_time < '2026-09-01'
{% endif %}
初回の実行ではCTASによってターゲットを作成し、INSERT OVERWRITEは実行しません。このため、この例ではpartition_by_initを使用して必要なターゲットパーティションをすべて作成します。ターゲットの作成後は、overwrite_partitionsが上書き範囲を制御します。
- 省略した場合: テーブル全体を上書きします。
- パーティション名のリスト: 指定したパーティションのみを上書きします。
'*': バッチが対象とするパーティションをDorisが動的に上書きします。
動的なOLAP上書きでは、空のバッチからクリアすべきパーティションを推論できません。ソースのバッチが空でもターゲットデータを削除する必要がある場合は、パーティション名を指定してください。
Icebergのターゲットでは、overwrite_partitionsを省略すると、パーティション化されたテーブルでネイティブの動的上書きが使用されます。一つの静的な範囲を置き換えるには、overwrite_partitions={'dt': '2025-01-25'}のように、空でない列と値のマッピングを指定します。モデルは、その範囲のすべての行を返す必要があります。パーティション化されていないターゲットでは、モデルは置き換え後の全データを返す必要があり、結果が空ならテーブルがクリアされます。静的な範囲でも、結果が空ならその範囲がクリアされます。NULLを含むマッピングではネイティブMERGEを使用し、Iceberg V2以降が必要です。
マイクロバッチ
microbatchはdbt Coreのバッチコンテキストを使用します。次の例はInternal CatalogのOLAPテーブルをターゲットにし、各UTCウィンドウに正確に対応する一つのRANGEパーティションを上書きします。直接参照する各上流リソースとターゲットモデルにevent_timeフィールドを設定してください。上記の最初のモデルの例で作成したraw.ordersソースについては、既存のテーブル宣言に次の設定を追加します。
sources:
- name: raw
schema: raw
tables:
- name: orders
config:
event_time: order_time
{{
config(
materialized='incremental',
incremental_strategy='microbatch',
event_time='event_time',
batch_size='day',
begin=modules.datetime.datetime(2026, 8, 1, 0, 0, 0),
duplicate_key=['event_id', 'event_time'],
partition_by=['event_time'],
partition_type='RANGE',
distributed_by=['event_id'],
buckets=16,
properties={'replication_num': '1'}
)
}}
select
order_id as event_id,
order_time as event_time,
status as event_type
from {{ source('raw', 'orders') }}
Internal CatalogのOLAPターゲットでは、event_timeは引用符で囲まない列名で、partition_byと同じ単一列を指す必要があります。batch_sizeはhour、day、month、yearに対応します。ターゲットはDuplicate Keyテーブルです。duplicate_keyはDorisのテーブルキーを設定するもので、dbtのunique_keyではありません。unique_key、overwrite_partitions、partition_by_initは設定しないでください。アダプターが現在のバッチからパーティション境界と上書き対象を導き出します。上流のref()またはsource()にもevent_timeが設定されていれば、dbt Coreはその入力を現在のウィンドウに絞り込みます。Microbatchの各バッチは順次実行され、空のバッチでも該当パーティションを上書きするため、そのウィンドウから削除された行もクリアされます。
次のプロパティを設定し、アダプターの検証に合格する場合、Internal CatalogのMicrobatchでDorisのDynamic Partitionも使用できます。
dynamic_partition.enable='true'。dynamic_partition.time_unitがbatch_sizeと一致し、dynamic_partition.time_zoneがUTC、Etc/UTC、+00:00のいずれかであること。dynamic_partition.create_history_partition='true'であり、dynamic_partition.startまたはdynamic_partition.history_partition_numのいずれかを指定すること。dynamic_partition.prefixが有効な識別子であり、dynamic_partition.endが正の整数であること。- 月単位のバッチでは、
dynamic_partition.start_day_of_monthを設定する場合、その値が1であること。
DorisではDynamic Partitionテーブルの作成時にdynamic_partition.bucketsの値も正である必要があります。このプロパティはアダプターではなくDorisが検証します。保持期間には、begin、lookback、手動でのバックフィル範囲が含まれている必要があります。既存のターゲットについては、アダプターが検証する物理プロパティがモデル設定と一致する必要があります。一致しない場合は設定を合わせるか、--full-refreshを実行してください。
append、merge、パーティション単位のinsert_overwriteでは、アダプターは新規または変更されたソース行を自動検出しません。モデルSQLが意図したバッチを返す必要があります。テーブル全体を対象にしたinsert_overwriteでは、ターゲットの全データを返す必要があります。各mergeバッチ内のunique_keyの値は一意でなければなりません。dbt Coreはevent_timeを設定した上流リソースからMicrobatch用のフィルタリングを行うため、Microbatchモデルでis_incremental()は不要です。キー、パーティション、または物理レイアウトに互換性のない変更を加えた後は、dbt run --full-refreshを実行してください。IcebergのMicrobatchでは、unique_key、overwrite_partitions（列と値のマッピングを含む）、partition_by_init、OLAPのDynamic Partitionプロパティを省略します。partition_byを設定する場合はpartition_type='LIST'を指定します。空のウィンドウも含め、各UTCウィンドウはネイティブMERGEで置き換えられます。
非同期マテリアライズドビュー
dbtユーザーは標準のmaterialized='materialized_view'を設定します。アダプターはDorisの非同期マテリアライズドビューを使用して実装します。
このマテリアライゼーションに対するFEのバージョン判定では、Doris 2.xの2.1.5以降、Doris 3.x（3.0.0を除く）、Doris 4.x以降、およびdoris-0.0.0-<git-sha>として報告される識別可能なソースビルドが許可されます。ソースビルドは開発テスト専用です。これらはコード上で受け入れる条件であり、許可されたすべてのバージョンで互換性テストに合格した証拠ではありません。
{{
config(
materialized='materialized_view',
build_mode='immediate',
refresh_method='auto',
refresh_trigger='manual',
duplicate_key=['order_date'],
distribution_type='hash',
distributed_by=['order_date'],
buckets=8,
wait_for_refresh=true,
properties={'replication_num': '1'}
)
}}
select order_date, sales_amount
from {{ ref('fct_daily_sales') }}
| 設定 | デフォルト | 対応する値または形式 |
|---|---|---|
build_mode | immediate | immediate, deferred |
refresh_method | auto | autoではDorisが更新範囲を決定し、completeではすべてのパーティションを更新します |
refresh_trigger | manual | manual, schedule, commit |
refresh_schedule | - | scheduleで必須。正のinterval、minute/hour/day/weekのいずれかの単位、省略可能なstart_timeを含みます |
wait_for_refresh | true | BUILD IMMEDIATEによる初回の構築、またはアダプターが送信した手動更新の完了を待つかどうか |
refresh_wait_timeout | 300 | 更新タスクのタイムアウト（秒）。正の整数 |
refresh_poll_interval | 1 | 更新タスクのポーリング間隔（秒）。タイムアウト以下の正の整数 |
duplicate_key | - | 非同期マテリアライズドビューのDuplicate Key列。文字列またはリスト |
partition_by | - | Dorisのパーティション列またはパーティション関数式を一つ指定。文字列または単一要素のリスト |
distribution_type | 自動選択 | hash（distributed_byを設定した場合）、それ以外はrandom |
distributed_by | - | ハッシュ分散の列 |
buckets | auto | 正の整数またはauto |
replication_num | - | レプリカ数を簡単に指定するための設定。properties内の同じキーより優先されます |
properties | {} | Doris非同期マテリアライズドビューのプロパティ |
on_configuration_change | apply | apply, continue, fail |
BUILD IMMEDIATEは、定義の作成時または置き換え時に初回の構築を開始します。アダプターは一時的な名前で新しい定義を作成し、デフォルトでは初回の構築を待ってからターゲット名で公開します。追加のREFRESH MATERIALIZED VIEWは送信しません。BUILD DEFERREDは初回の構築を行わずに定義を作成します。
定義が変わっていない場合、更新動作はrefresh_triggerだけで制御されます。現時点でrefresh_on_run設定はありません。
manualでは、REFRESH MATERIALIZED VIEW ... AUTO|COMPLETEを後続のdbt runでモデルが選択されたときに送信し、デフォルトでは成功するまで待ちます。scheduleとcommitでは、後続のdbt実行から更新を送信しません。Dorisがスケジュールまたはベーステーブルへのコミットを契機に更新します。
たとえば、日次スケジュールは次のように設定します。
{{
config(
materialized='materialized_view',
build_mode='deferred',
refresh_method='auto',
refresh_trigger='schedule',
refresh_schedule={
'interval': 1,
'unit': 'day'
},
properties={'replication_num': '1'}
)
}}
select order_date, sales_amount
from {{ ref('fct_daily_sales') }}
wait_for_refresh=falseはタスクのポーリングを無効にしますが、更新リクエストは抑止しません。dbt側の待機がタイムアウトしても、送信済みのDorisタスクはキャンセルされません。アダプターは、送信前後のDoris MVタスクIDを比較して新しいタスクを識別します。そのため、同じターゲットに対する手動更新を並行して実行しないでください。
モデルSQLまたはDDL設定が変わった場合、デフォルトのon_configuration_change=applyは一時的な非同期マテリアライズドビューを構築し、既存の定義をアトミックに置き換えます。continueは手動更新を送信せずに既存の定義を維持し、failは実行を停止します。--full-refreshは定義の再デプロイを強制します。
その他のdbt機能
| 機能 | 対応状況 |
|---|---|
| Seed | seeds/にバージョン管理された小さなCSVファイルを配置し、dbt seedを実行します。型推論、column_types、refに対応します |
| Snapshot | checkまたはtimestamp戦略をsnapshots/で定義し、dbt snapshotを実行します。ハードデリートとスキーマの変更に対応します。Internal Catalogでの置き換えはアトミックですが、Icebergではバックアップと名前変更を使用します |
| データテスト | YAMLでGeneric Testを設定するか、tests/にSingular Testを記述します。ephemeralとstore_failuresの処理経路に対応します |
| ユニットテスト | モデルのYAMLに行を直接記述するかCSVのフィクスチャーを指定し、dbt test --select test_type:unitを実行します |
| モデルコントラクト | contract.enforced: trueを設定し、Table、View、Incrementalモデルの列名と型を宣言します。constraints宣言は拒否され、データテストはデータベースの制約を作成しません |
| 永続化されたドキュメント | persist_docsを有効にして、主要なリレーション型にリレーションと列のコメントを設定します |
| ソースの鮮度 | loaded_at_fieldまたはloaded_at_queryをfreshnessとともに設定し、dbt source freshnessを実行します |
| 権限付与 | 標準のgrants設定をDorisのuserおよびuser@hostのテーブル権限に使用します。ロールをプリンシパルにすることは現時点でできません |
| フック | pre_hookとpost_hookを使用して、マテリアライゼーションの前後にDoris SQLを実行します。フックの副作用はトランザクションでロールバックされません |
| ドキュメントとリネージ | 説明を記述してdbt docs generateを実行し、Internal Catalogの非同期マテリアライズドビューを含むInternal CatalogとExternal Catalogのリレーション・列のメタデータを生成します |
Seedは国コードやステータスの対応表など、バージョン管理される小規模なデータに適しています。大量の業務データの取り込みには適しません。大きなファイルにはDoris Stream Load、Broker Load、またはDoris Connectorを使用してください。
Snapshotのソースにあるunique_keyの値はNULL以外で、各バッチ内で一意でなければなりません。timestamp戦略では、updated_atもNULL以外で、そのキーの現在の履歴バージョンより前の値にすることはできません。
テーブルと列の説明を永続化するには、次のようにします。
models:
- name: fct_daily_sales
description: Daily sales summary
config:
persist_docs:
relation: true
columns: true
columns:
- name: order_date
description: Order date
- name: sales_amount
description: Daily sales amount
ソースの鮮度
ソースの鮮度は、タイムスタンプを明示的に指定する二つの方法に対応します。一般的なタイムスタンプ列を使う場合はloaded_at_fieldを使用します。
sources:
- name: raw
schema: raw
tables:
- name: orders
config:
loaded_at_field: updated_at
freshness:
warn_after: {count: 1, period: hour}
error_after: {count: 2, period: hour}
鮮度を判定するタイムスタンプにカスタムSQLまたは集約が必要な場合は、loaded_at_queryを使用します。
sources:
- name: raw
schema: raw
tables:
- name: orders
config:
loaded_at_query: |
select max(updated_at)
from {{ this }}
where is_valid = 1
freshness:
warn_after: {count: 1, period: hour}
error_after: {count: 2, period: hour}
loaded_at_fieldを使用する場合は、鮮度評価の対象行を絞り込むため、filterをfreshnessの下に設定します。たとえばfilter: is_valid = 1を指定します。loaded_at_fieldとloaded_at_queryを同時に設定しないでください。現時点でアダプターはDorisのテーブルメタデータからロード時刻を推論しません。フィールドまたはクエリを明示的に指定してください。
アダプターは現在時刻としてutc_timestamp()を使用します。これはタイムゾーン情報のない鮮度判定用タイムスタンプをUTCとして評価するdbt Coreの慣例に沿っています。
権限付与
宣言的な権限付与ではDorisユーザーをプリンシパルとして使用します。
models:
doris_demo:
+grants:
select:
- analyst
- reporter@10.0.0.%
@hostを含まない名前はusername@%として扱われます。権限の対応は、selectがSELECT_PRIV、insertがLOAD_PRIV、alterがALTER_PRIV、createがCREATE_PRIV、dropがDROP_PRIV、show_viewがSHOW_VIEW_PRIVです。Dorisのロールは現時点でgrantsに対応していないため、具体的なユーザーを設定してください。dbtの実行アカウントには、プリンシパルの確認とターゲットオブジェクトの権限調整ができる必要があります。
よく使うコマンド
| コマンド | 目的 |
|---|---|
dbt debug | プロジェクト設定とDorisへの接続を検証します |
dbt parse | SQLを実行せずにプロジェクトを解析します |
dbt compile | モデルをコンパイルし、生成されたDoris SQLを確認します |
dbt run | 選択したモデルを構築します |
dbt test | データテストとユニットテストを実行します |
dbt build | 依存関係の順序に従ってSeed、モデル、Snapshot、テストを実行します |
dbt seed | SeedのCSVファイルをロードします |
dbt snapshot | Snapshotを更新します |
dbt source freshness | ソースの鮮度を評価します |
dbt docs generate | カタログ、ドキュメント、リネージのメタデータを生成します |
よく使うセレクター:
# Run one model
dbt run --select fct_daily_sales
# Run the model and all upstream dependencies
dbt build --select +fct_daily_sales
# Run the model and all downstream dependencies
dbt build --select fct_daily_sales+
# Fully rebuild an incremental model
dbt run --select fct_orders --full-refresh
現在の制限事項
- 現時点でアダプターは、Aggregate Keyテーブルのモデリングとセカンダリインデックスの設定に対応していません。
- External Catalogへの書き込み動作はDorisコネクターに依存します。Catalogs V2とコネクター固有の書き込み保証には対応していません。
- 現時点でアダプターは、SSLの設定、タイムアウトとリトライ、複数FEのフェイルオーバー、サーバー側でのキャンセル、完全なクエリテレメトリーを実装していません。
- 一部のTable、View、MVの型変更では、無停止で切り替わるのではなく、正規の名前が短時間利用できなくなります。
トラブルシューティング
dbt debugで接続できない
hostとportが到達可能なFEのMySQLクエリポートを指していることを確認し、ユーザー名、パスワード、ネットワークポリシーを確認してください。クエリポートのデフォルトは9030です。8030は通常、FEのHTTPポートです。
Catalogまたはデータベースが見つからない
v1.1.1では、プロファイルとソースのdatabaseの値はDoris Catalogを、schemaはその中のデータベースを表します。Internal Catalogを使用する場合はdatabaseを省略します。古いプロジェクトでソースのdatabaseをDorisデータベースの指定に使用していた場合は、その値をschemaに移してください。External Catalogが既に存在し、dbtの実行アカウントから参照できることを確認してください。