ガイド

ClickHouse対応

このセクションでは、dbt と ClickHouse アダプターのセットアップ方法に関するガイドに加え、一般公開されている IMDB データセットを使用して ClickHouse で dbt を利用する例を紹介します。この例では、次の手順を扱います。

dbt プロジェクトを作成し、ClickHouse アダプターをセットアップする。
モデルを定義する。
モデルを更新する。
incremental モデルを作成する。
スナップショットモデルを作成する。
materialized view を使用する。

これらのガイドは、ドキュメント、機能と設定、およびマテリアライゼーションリファレンスとあわせて利用することを想定しています。

セットアップ

環境を準備するには、dbt と ClickHouse アダプターのセットアップセクションの手順に従ってください。

重要: 以下の内容は Python 3.9 でテストされています。

ClickHouse を準備する

dbt は、リレーショナル性の高いデータのモデリングで特に威力を発揮します。例として、以下のリレーショナルスキーマを持つ小規模な IMDB データセットを用意しています。このデータセットは relational dataset repository から取得したものです。dbt で一般的に扱うスキーマと比べると非常に単純ですが、手頃なサンプルとして適しています。

これらのテーブルの一部を、以下に示すように使用します。

次のテーブルを作成します。

CREATE DATABASE imdb;

CREATE TABLE imdb.actors
(
    id         UInt32,
    first_name String,
    last_name  String,
    gender     FixedString(1)
) ENGINE = MergeTree ORDER BY (id, first_name, last_name, gender);

CREATE TABLE imdb.directors
(
    id         UInt32,
    first_name String,
    last_name  String
) ENGINE = MergeTree ORDER BY (id, first_name, last_name);

CREATE TABLE imdb.genres
(
    movie_id UInt32,
    genre    String
) ENGINE = MergeTree ORDER BY (movie_id, genre);

CREATE TABLE imdb.movie_directors
(
    director_id UInt32,
    movie_id    UInt64
) ENGINE = MergeTree ORDER BY (director_id, movie_id);

CREATE TABLE imdb.movies
(
    id   UInt32,
    name String,
    year UInt32,
    rank Float32 DEFAULT 0
) ENGINE = MergeTree ORDER BY (id, name, year);

CREATE TABLE imdb.roles
(
    actor_id   UInt32,
    movie_id   UInt32,
    role       String,
    created_at DateTime DEFAULT now()
) ENGINE = MergeTree ORDER BY (actor_id, movie_id);

s3 関数を使用して、公開エンドポイントからソースデータを読み込み、データを挿入します。次のコマンドを実行してテーブルにデータを投入してください。

INSERT INTO imdb.actors
SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/imdb/imdb_ijs_actors.tsv.gz', NOSIGN,
'TSVWithNames');

INSERT INTO imdb.directors
SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/imdb/imdb_ijs_directors.tsv.gz', NOSIGN,
'TSVWithNames');

INSERT INTO imdb.genres
SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/imdb/imdb_ijs_movies_genres.tsv.gz', NOSIGN,
'TSVWithNames');

INSERT INTO imdb.movie_directors
SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/imdb/imdb_ijs_movies_directors.tsv.gz', NOSIGN,
        'TSVWithNames');

INSERT INTO imdb.movies
SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/imdb/imdb_ijs_movies.tsv.gz', NOSIGN,
'TSVWithNames');

INSERT INTO imdb.roles(actor_id, movie_id, role)
SELECT actor_id, movie_id, role
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/imdb/imdb_ijs_roles.tsv.gz', NOSIGN,
'TSVWithNames');

これらの実行時間は帯域幅によって異なる場合がありますが、いずれも完了まで数秒しかかからないはずです。以下のクエリを実行して、映画への出演数が多い順に各俳優の集計結果を算出し、データが正常に読み込まれたことを確認します。

SELECT id,
       any(actor_name)          AS name,
       uniqExact(movie_id)    AS num_movies,
       avg(rank)                AS avg_rank,
       uniqExact(genre)         AS unique_genres,
       uniqExact(director_name) AS uniq_directors,
       max(created_at)          AS updated_at
FROM (
         SELECT imdb.actors.id  AS id,
                concat(imdb.actors.first_name, ' ', imdb.actors.last_name)  AS actor_name,
                imdb.movies.id AS movie_id,
                imdb.movies.rank AS rank,
                genre,
                concat(imdb.directors.first_name, ' ', imdb.directors.last_name) AS director_name,
                created_at
         FROM imdb.actors
                  JOIN imdb.roles ON imdb.roles.actor_id = imdb.actors.id
                  LEFT OUTER JOIN imdb.movies ON imdb.movies.id = imdb.roles.movie_id
                  LEFT OUTER JOIN imdb.genres ON imdb.genres.movie_id = imdb.movies.id
                  LEFT OUTER JOIN imdb.movie_directors ON imdb.movie_directors.movie_id = imdb.movies.id
                  LEFT OUTER JOIN imdb.directors ON imdb.directors.id = imdb.movie_directors.director_id
         )
GROUP BY id
ORDER BY num_movies DESC
LIMIT 5;

レスポンスは次のようになります。

+------+------------+----------+------------------+-------------+--------------+-------------------+
|id    |name        |num_movies|avg_rank          |unique_genres|uniq_directors|updated_at         |
+------+------------+----------+------------------+-------------+--------------+-------------------+
|45332 |Mel Blanc   |832       |6.175853582979779 |18           |84            |2022-04-26 14:01:45|
|621468|Bess Flowers|659       |5.57727638854796  |19           |293           |2022-04-26 14:01:46|
|372839|Lee Phelps  |527       |5.032976449684617 |18           |261           |2022-04-26 14:01:46|
|283127|Tom London  |525       |2.8721716524875673|17           |203           |2022-04-26 14:01:46|
|356804|Bud Osborne |515       |2.0389507108727773|15           |149           |2022-04-26 14:01:46|
+------+------------+----------+------------------+-------------+--------------+-------------------+

以降のガイドでは、このクエリをモデルに変換し、dbt のビューおよびテーブルとして ClickHouse にマテリアライズします。

ClickHouse への接続

dbt プロジェクトを作成します。この例では、imdb ソースにちなんで名前を付けます。プロンプトが表示されたら、データベースソースとして clickhouse を選択します。

clickhouse-user@clickhouse:~$ dbt init imdb

16:52:40  Running with dbt=1.1.0
Which database would you like to use?
[1] clickhouse

(Don't see the one you want? https://docs.getdbt.com/docs/available-adapters)

Enter a number: 1
16:53:21  No sample profile found for clickhouse.
16:53:21
Your new dbt project "imdb" was created!

For more information on how to configure the profiles.yml file,
please consult the dbt documentation here:

https://docs.getdbt.com/docs/configure-your-profile

プロジェクトフォルダーに cd します。
```
cd imdb
```
この時点で、お好みのテキストエディターが必要です。以下の例では、広く使われている VS Code を使用します。IMDB ディレクトリを開くと、yml ファイルと sql ファイルが一式表示されるはずです。
dbt_project.yml ファイルを更新し、最初のモデル actor_summary を指定して、プロファイルを clickhouse_imdb に設定します。
次に、dbt に ClickHouse インスタンスの接続情報を指定する必要があります。以下を ~/.dbt/profiles.yml に追加してください。
```
clickhouse_imdb:
  target: dev
  outputs:
    dev:
      type: clickhouse
      schema: imdb_dbt
      host: localhost
      port: 8123
      user: default
      password: ''
      secure: False
```
user と password は変更が必要である点に注意してください。利用可能な追加の設定はこちらに記載されています。

IMDB ディレクトリから dbt debug コマンドを実行し、dbt が ClickHouse に接続できるかどうかを確認します。

clickhouse-user@clickhouse:~/imdb$ dbt debug
17:33:53  Running with dbt=1.1.0
dbt version: 1.1.0
python version: 3.10.1
python path: /home/dale/.pyenv/versions/3.10.1/bin/python3.10
os info: Linux-5.13.0-10039-tuxedo-x86_64-with-glibc2.31
Using profiles.yml file at /home/dale/.dbt/profiles.yml
Using dbt_project.yml file at /opt/dbt/imdb/dbt_project.yml

Configuration:
profiles.yml file [OK found and valid]
dbt_project.yml file [OK found and valid]

Required dependencies:
- git [OK found]

Connection:
host: localhost
port: 8123
user: default
schema: imdb_dbt
secure: False
verify: False
Connection test: [OK connection ok]

All checks passed!

接続に成功したことを示す Connection test: [OK connection ok] が応答に含まれていることを確認してください。

シンプルなビューマテリアライゼーションの作成

ビューマテリアライゼーションを使用する場合、モデルは実行のたびに ClickHouse で CREATE VIEW AS ステートメントによってビューとして再構築されます。これにはデータを追加で保存する必要はありませんが、テーブルマテリアライゼーションよりクエリは遅くなります。

imdb フォルダ内の models/example ディレクトリを削除します:
```
clickhouse-user@clickhouse:~/imdb$ rm -rf models/example
```
models フォルダ内の actors に新しいファイルを作成します。ここでは、それぞれが actor モデルを表すファイルを作成します。
```
clickhouse-user@clickhouse:~/imdb$ mkdir models/actors
```

models/actors フォルダに schema.yml と actor_summary.sql を作成します。

clickhouse-user@clickhouse:~/imdb$ touch models/actors/actor_summary.sql
clickhouse-user@clickhouse:~/imdb$ touch models/actors/schema.yml

ファイル schema.yml でテーブルを定義します。これらは以降、マクロで利用できるようになります。 models/actors/schema.yml を次の内容に編集してください:

version: 2

sources:
- name: imdb
  tables:
  - name: directors
  - name: actors
  - name: roles
  - name: movies
  - name: genres
  - name: movie_directors

actors_summary.sql では、実際のモデルを定義します。config 関数では、このモデルを ClickHouse で view として materialize するよう指定している点にも注意してください。各テーブルは、schema.yml ファイルから関数 source を介して参照されます。たとえば source('imdb', 'movies') は、imdb データベース内の movies テーブルを参照します。models/actors/actors_summary.sql を編集して、次の内容にしてください:

{{ config(materialized='view') }}

with actor_summary as (
SELECT id,
    any(actor_name) as name,
    uniqExact(movie_id)    as num_movies,
    avg(rank)                as avg_rank,
    uniqExact(genre)         as genres,
    uniqExact(director_name) as directors,
    max(created_at) as updated_at
FROM (
        SELECT {{ source('imdb', 'actors') }}.id as id,
                concat({{ source('imdb', 'actors') }}.first_name, ' ', {{ source('imdb', 'actors') }}.last_name) as actor_name,
                {{ source('imdb', 'movies') }}.id as movie_id,
                {{ source('imdb', 'movies') }}.rank as rank,
                genre,
                concat({{ source('imdb', 'directors') }}.first_name, ' ', {{ source('imdb', 'directors') }}.last_name) as director_name,
                created_at
        FROM {{ source('imdb', 'actors') }}
                    JOIN {{ source('imdb', 'roles') }} ON {{ source('imdb', 'roles') }}.actor_id = {{ source('imdb', 'actors') }}.id
                    LEFT OUTER JOIN {{ source('imdb', 'movies') }} ON {{ source('imdb', 'movies') }}.id = {{ source('imdb', 'roles') }}.movie_id
                    LEFT OUTER JOIN {{ source('imdb', 'genres') }} ON {{ source('imdb', 'genres') }}.movie_id = {{ source('imdb', 'movies') }}.id
                    LEFT OUTER JOIN {{ source('imdb', 'movie_directors') }} ON {{ source('imdb', 'movie_directors') }}.movie_id = {{ source('imdb', 'movies') }}.id
                    LEFT OUTER JOIN {{ source('imdb', 'directors') }} ON {{ source('imdb', 'directors') }}.id = {{ source('imdb', 'movie_directors') }}.director_id
        )
GROUP BY id
)

select *
from actor_summary

最終的な actor_summary にカラム updated_at を含めている点に注目してください。これは後で増分マテリアライゼーションに利用します。

imdb ディレクトリで dbt run コマンドを実行します。

clickhouse-user@clickhouse:~/imdb$ dbt run
15:05:35  Running with dbt=1.1.0
15:05:35  Found 1 model, 0 tests, 1 snapshot, 0 analyses, 181 macros, 0 operations, 0 seed files, 6 sources, 0 exposures, 0 metrics
15:05:35
15:05:36  Concurrency: 1 threads (target='dev')
15:05:36
15:05:36  1 of 1 START view model imdb_dbt.actor_summary.................................. [RUN]
15:05:37  1 of 1 OK created view model imdb_dbt.actor_summary............................. [OK in 1.00s]
15:05:37
15:05:37  Finished running 1 view model in 1.97s.
15:05:37
15:05:37  Completed successfully
15:05:37
15:05:37  Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1

dbt は、指定どおりにこのモデルを ClickHouse のビューとして表現します。これで、このビューに直接クエリを実行できます。このビューは imdb_dbt データベースに作成されます。これは、clickhouse_imdb プロファイル配下の ~/.dbt/profiles.yml ファイルにあるスキーマパラメータによって決まります。

SHOW DATABASES;

+------------------+
|name              |
+------------------+
|INFORMATION_SCHEMA|
|default           |
|imdb              |
|imdb_dbt          |  <---dbtによって作成！
|information_schema|
|system            |
+------------------+

このビューに対してクエリを実行すると、よりシンプルな構文で先ほどのクエリと同じ結果を再現できます:

SELECT * FROM imdb_dbt.actor_summary ORDER BY num_movies DESC LIMIT 5;

+------+------------+----------+------------------+------+---------+-------------------+
|id    |name        |num_movies|avg_rank          |genres|directors|updated_at         |
+------+------------+----------+------------------+------+---------+-------------------+
|45332 |Mel Blanc   |832       |6.175853582979779 |18    |84       |2022-04-26 15:26:55|
|621468|Bess Flowers|659       |5.57727638854796  |19    |293      |2022-04-26 15:26:57|
|372839|Lee Phelps  |527       |5.032976449684617 |18    |261      |2022-04-26 15:26:56|
|283127|Tom London  |525       |2.8721716524875673|17    |203      |2022-04-26 15:26:56|
|356804|Bud Osborne |515       |2.0389507108727773|15    |149      |2022-04-26 15:26:56|
+------+------------+----------+------------------+------+---------+-------------------+

テーブルとしてのマテリアライゼーションの作成

前の例では、モデルはビューとしてマテリアライズされました。これは一部のクエリには十分なパフォーマンスを提供する場合もありますが、より複雑なSELECTや頻繁に実行されるクエリでは、テーブルとしてマテリアライズしたほうが適していることがあります。このマテリアライゼーションは、BIツールからクエリされるモデルに適しており、ユーザーにより高速な操作性を提供できます。これにより、クエリ結果は新しいテーブルとして保存され、その分のストレージオーバーヘッドが発生します。つまり、実質的には INSERT TO SELECT が実行されます。このテーブルは毎回再構築されるため、増分ではない点に注意してください。そのため、結果セットが大きい場合は実行時間が長くなる可能性があります。詳細は dbt Limitations を参照してください。

actors_summary.sql ファイルを修正し、materialized パラメータが table に設定されるようにします。ORDER BY がどのように定義されているか、また MergeTree テーブルエンジンを使用していることに注目してください。
```
{{ config(order_by='(updated_at, id, name)', engine='MergeTree()', materialized='table') }}
```

imdb ディレクトリで dbt run コマンドを実行します。この処理にはやや時間がかかる場合があり、ほとんどのマシンでは約 10 秒です。

clickhouse-user@clickhouse:~/imdb$ dbt run
15:13:27  Running with dbt=1.1.0
15:13:27  Found 1 model, 0 tests, 1 snapshot, 0 analyses, 181 macros, 0 operations, 0 seed files, 6 sources, 0 exposures, 0 metrics
15:13:27
15:13:28  Concurrency: 1 threads (target='dev')
15:13:28
15:13:28  1 of 1 START table model imdb_dbt.actor_summary................................. [RUN]
15:13:37  1 of 1 OK created table model imdb_dbt.actor_summary............................ [OK in 9.22s]
15:13:37
15:13:37  Finished running 1 table model in 10.20s.
15:13:37
15:13:37  Completed successfully
15:13:37
15:13:37  Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1

imdb_dbt.actor_summary テーブルが作成されたことを確認します。

SHOW CREATE TABLE imdb_dbt.actor_summary;

適切なデータ型を持つテーブルが表示されるはずです。

+----------------------------------------
|statement
+----------------------------------------
|CREATE TABLE imdb_dbt.actor_summary
|(
|`id` UInt32,
|`first_name` String,
|`last_name` String,
|`num_movies` UInt64,
|`updated_at` DateTime
|)
|ENGINE = MergeTree
|ORDER BY (id, first_name, last_name)
+----------------------------------------

このテーブルの結果が前のレスポンスと一致していることを確認します。モデルがテーブルになったことで、レスポンス時間が大幅に改善されていることがわかります。

SELECT * FROM imdb_dbt.actor_summary ORDER BY num_movies DESC LIMIT 5;

+------+------------+----------+------------------+------+---------+-------------------+
|id    |name        |num_movies|avg_rank          |genres|directors|updated_at         |
+------+------------+----------+------------------+------+---------+-------------------+
|45332 |Mel Blanc   |832       |6.175853582979779 |18    |84       |2022-04-26 15:26:55|
|621468|Bess Flowers|659       |5.57727638854796  |19    |293      |2022-04-26 15:26:57|
|372839|Lee Phelps  |527       |5.032976449684617 |18    |261      |2022-04-26 15:26:56|
|283127|Tom London  |525       |2.8721716524875673|17    |203      |2022-04-26 15:26:56|
|356804|Bud Osborne |515       |2.0389507108727773|15    |149      |2022-04-26 15:26:56|
+------+------------+----------+------------------+------+---------+-------------------+

このモデルに対して、他のクエリも自由に実行してみてください。たとえば、出演回数が 5 回を超える俳優のうち、平均評価が最も高いのは誰でしょうか。

SELECT * FROM imdb_dbt.actor_summary WHERE num_movies > 5 ORDER BY avg_rank  DESC LIMIT 10;

インクリメンタルマテリアライゼーションの作成

前の例では、モデルをマテリアライズするためのテーブルを作成しました。このテーブルは、dbt の実行のたびに再構築されます。結果セットが大きい場合や変換が複雑な場合、これは現実的でなく、コストも非常に高くなる可能性があります。この課題に対処してビルド時間を短縮するために、dbt にはインクリメンタルマテリアライゼーションが用意されています。これにより、dbt は前回の実行以降にテーブル内のレコードを挿入または更新できるため、イベント形式のデータに適しています。内部的には、まず更新されたすべてのレコードを含む一時テーブルが作成され、その後、変更されていないすべてのレコードと更新済みのレコードが新しいターゲットテーブルに挿入されます。そのため、大きな結果セットでは、テーブルモデルと同様の制約があります。

大規模なセットでこれらの制約を回避するために、アダプターは inserts_only モードをサポートしています。このモードでは、一時テーブルを作成せずに、すべての更新がターゲットテーブルに挿入されます (詳細は後述します) 。

この例を示すために、俳優「Clicky McClickHouse」を追加します。彼は驚異の 910 本の映画に出演し、Mel Blanc をも上回る出演本数になります。

まず、モデルを incremental 型に変更します。この変更には、次の対応が必要です。

unique_key - アダプターが行を一意に識別できるようにするには、unique_key を指定する必要があります。この場合は、クエリ内の id フィールドで十分です。これにより、マテリアライズされたテーブル内で行の重複が発生しないようにできます。一意性制約の詳細については、こちらを参照してください。
Incremental filter - また、増分実行時にどの行が変更されたかを dbt がどのように識別するかも指定する必要があります。これは、差分を表す式を指定することで実現します。通常、イベントデータでは timestamp を使うため、ここでは updated_at timestamp フィールドを使用します。このカラムは、行が挿入されると既定で now() の値が設定されるため、新しい行を特定できます。さらに、新しいアクターが追加されるケースも検出する必要があります。既存のマテリアライズされたテーブルを表す {{this}} 変数を使うと、式は where id > (select max(id) from {{ this }}) or updated_at > (select max(updated_at) from {{this}}) になります。これを {% if is_incremental() %} 条件内に埋め込むことで、増分実行時にのみ使用され、テーブルの初回作成時には使用されないようにしています。増分モデルで行をフィルタリングする詳細については、dbt ドキュメントのこちらの説明を参照してください。

actor_summary.sql ファイルを以下のように更新します。

{{ config(order_by='(updated_at, id, name)', engine='MergeTree()', materialized='incremental', unique_key='id') }}
with actor_summary as (
    SELECT id,
        any(actor_name) as name,
        uniqExact(movie_id)    as num_movies,
        avg(rank)                as avg_rank,
        uniqExact(genre)         as genres,
        uniqExact(director_name) as directors,
        max(created_at) as updated_at
    FROM (
        SELECT {{ source('imdb', 'actors') }}.id as id,
            concat({{ source('imdb', 'actors') }}.first_name, ' ', {{ source('imdb', 'actors') }}.last_name) as actor_name,
            {{ source('imdb', 'movies') }}.id as movie_id,
            {{ source('imdb', 'movies') }}.rank as rank,
            genre,
            concat({{ source('imdb', 'directors') }}.first_name, ' ', {{ source('imdb', 'directors') }}.last_name) as director_name,
            created_at
    FROM {{ source('imdb', 'actors') }}
        JOIN {{ source('imdb', 'roles') }} ON {{ source('imdb', 'roles') }}.actor_id = {{ source('imdb', 'actors') }}.id
        LEFT OUTER JOIN {{ source('imdb', 'movies') }} ON {{ source('imdb', 'movies') }}.id = {{ source('imdb', 'roles') }}.movie_id
        LEFT OUTER JOIN {{ source('imdb', 'genres') }} ON {{ source('imdb', 'genres') }}.movie_id = {{ source('imdb', 'movies') }}.id
        LEFT OUTER JOIN {{ source('imdb', 'movie_directors') }} ON {{ source('imdb', 'movie_directors') }}.movie_id = {{ source('imdb', 'movies') }}.id
        LEFT OUTER JOIN {{ source('imdb', 'directors') }} ON {{ source('imdb', 'directors') }}.id = {{ source('imdb', 'movie_directors') }}.director_id
    )
    GROUP BY id
)
select *
from actor_summary

{% if is_incremental() %}

-- このフィルターは増分実行時にのみ適用されます
where id > (select max(id) from {{ this }}) or updated_at > (select max(updated_at) from {{this}})

{% endif %}

このモデルは、roles テーブルと actors テーブルに対する更新と追加にのみ応答する点に注意してください。すべてのテーブルに対応させるには、このモデルを複数のサブモデルに分割し、それぞれに独自の増分条件を設定することをおすすめします。こうしたモデルは、さらに相互に参照したり接続したりできます。モデルの相互参照の詳細については、こちらを参照してください。

dbt run を実行し、生成されたテーブルの結果を確認します:

clickhouse-user@clickhouse:~/imdb$  dbt run
15:33:34  Running with dbt=1.1.0
15:33:34  Found 1 model, 0 tests, 1 snapshot, 0 analyses, 181 macros, 0 operations, 0 seed files, 6 sources, 0 exposures, 0 metrics
15:33:34
15:33:35  Concurrency: 1 threads (target='dev')
15:33:35
15:33:35  1 of 1 START incremental model imdb_dbt.actor_summary........................... [RUN]
15:33:41  1 of 1 OK created incremental model imdb_dbt.actor_summary...................... [OK in 6.33s]
15:33:41
15:33:41  Finished running 1 incremental model in 7.30s.
15:33:41
15:33:41  Completed successfully
15:33:41
15:33:41  Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1

SELECT * FROM imdb_dbt.actor_summary ORDER BY num_movies DESC LIMIT 5;

+------+------------+----------+------------------+------+---------+-------------------+
|id    |name        |num_movies|avg_rank          |genres|directors|updated_at         |
+------+------------+----------+------------------+------+---------+-------------------+
|45332 |Mel Blanc   |832       |6.175853582979779 |18    |84       |2022-04-26 15:26:55|
|621468|Bess Flowers|659       |5.57727638854796  |19    |293      |2022-04-26 15:26:57|
|372839|Lee Phelps  |527       |5.032976449684617 |18    |261      |2022-04-26 15:26:56|
|283127|Tom London  |525       |2.8721716524875673|17    |203      |2022-04-26 15:26:56|
|356804|Bud Osborne |515       |2.0389507108727773|15    |149      |2022-04-26 15:26:56|
+------+------------+----------+------------------+------+---------+-------------------+

ここで、増分更新を示すために、モデルにデータを追加します。actors テーブルに、俳優 "Clicky McClickHouse" を追加してください:
```
INSERT INTO imdb.actors VALUES (845466, 'Clicky', 'McClickHouse', 'M');
```

"Clicky"をランダムに選んだ910本の映画に出演させましょう:

INSERT INTO imdb.roles
SELECT now() as created_at, 845466 as actor_id, id as movie_id, 'Himself' as role
FROM imdb.movies
LIMIT 910 OFFSET 10000;

基になるソーステーブルを直接クエリし、dbtモデルを介さずに、彼が現在、本当に最も出演回数の多い俳優であることを確認します:

SELECT id,
    any(actor_name)          as name,
    uniqExact(movie_id)    as num_movies,
    avg(rank)                as avg_rank,
    uniqExact(genre)         as unique_genres,
    uniqExact(director_name) as uniq_directors,
    max(created_at)          as updated_at
FROM (
        SELECT imdb.actors.id                                                   as id,
                concat(imdb.actors.first_name, ' ', imdb.actors.last_name)       as actor_name,
                imdb.movies.id as movie_id,
                imdb.movies.rank                                                 as rank,
                genre,
                concat(imdb.directors.first_name, ' ', imdb.directors.last_name) as director_name,
                created_at
        FROM imdb.actors
                JOIN imdb.roles ON imdb.roles.actor_id = imdb.actors.id
                LEFT OUTER JOIN imdb.movies ON imdb.movies.id = imdb.roles.movie_id
                LEFT OUTER JOIN imdb.genres ON imdb.genres.movie_id = imdb.movies.id
                LEFT OUTER JOIN imdb.movie_directors ON imdb.movie_directors.movie_id = imdb.movies.id
                LEFT OUTER JOIN imdb.directors ON imdb.directors.id = imdb.movie_directors.director_id
        )
GROUP BY id
ORDER BY num_movies DESC
LIMIT 2;

+------+-------------------+----------+------------------+------+---------+-------------------+
|id    |name               |num_movies|avg_rank          |genres|directors|updated_at         |
+------+-------------------+----------+------------------+------+---------+-------------------+
|845466|Clicky McClickHouse|910       |1.4687938697032283|21    |662      |2022-04-26 16:20:36|
|45332 |Mel Blanc          |909       |5.7884792542982515|19    |148      |2022-04-26 16:17:42|
+------+-------------------+----------+------------------+------+---------+-------------------+

dbt run を実行し、モデルが更新されており、上記の結果と一致していることを確認します:

clickhouse-user@clickhouse:~/imdb$  dbt run
16:12:16  Running with dbt=1.1.0
16:12:16  Found 1 model, 0 tests, 1 snapshot, 0 analyses, 181 macros, 0 operations, 0 seed files, 6 sources, 0 exposures, 0 metrics
16:12:16
16:12:17  Concurrency: 1 threads (target='dev')
16:12:17
16:12:17  1 of 1 START incremental model imdb_dbt.actor_summary........................... [RUN]
16:12:24  1 of 1 OK created incremental model imdb_dbt.actor_summary...................... [OK in 6.82s]
16:12:24
16:12:24  Finished running 1 incremental model in 7.79s.
16:12:24
16:12:24  Completed successfully
16:12:24
16:12:24  Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1

SELECT * FROM imdb_dbt.actor_summary ORDER BY num_movies DESC LIMIT 2;

+------+-------------------+----------+------------------+------+---------+-------------------+
|id    |name               |num_movies|avg_rank          |genres|directors|updated_at         |
+------+-------------------+----------+------------------+------+---------+-------------------+
|845466|Clicky McClickHouse|910       |1.4687938697032283|21    |662      |2022-04-26 16:20:36|
|45332 |Mel Blanc          |909       |5.7884792542982515|19    |148      |2022-04-26 16:17:42|
+------+-------------------+----------+------------------+------+---------+-------------------+

内部

上記のインクリメンタル更新を実現するために実行されたステートメントは、ClickHouseのクエリログを照会することで特定できます。

SELECT event_time, query  FROM system.query_log WHERE type='QueryStart' AND query LIKE '%dbt%'
AND event_time > subtractMinutes(now(), 15) ORDER BY event_time LIMIT 100;

上記のクエリは、実行する期間に合わせて調整してください。結果の確認はユーザーに委ねますが、インクリメンタル更新を行うためにアダプターが使用する一般的な戦略を以下に示します。

アダプターは一時テーブル actor_sumary__dbt_tmp を作成します。変更された行はこのテーブルに書き込まれます。
次に、新しいテーブル actor_summary_new, が作成されます。続いて、古いテーブルの行が古いテーブルから新しいテーブルへ書き込まれますが、その際、一時テーブルに行 ID が存在しないことを確認します。これにより、更新と重複を効果的に処理できます。
一時テーブルの結果が、新しい actor_summary テーブルに書き込まれます。
最後に、新しいテーブルは EXCHANGE TABLES ステートメントを介して古いバージョンとアトミックに入れ替えられます。その後、古いテーブルと一時テーブルは削除されます。

これは以下の図のようになります。

この戦略は、非常に大規模なモデルでは課題が生じる可能性があります。詳細については、制限事項を参照してください。

Append 戦略 (insert のみモード)

増分モデルで大規模なデータセットを扱う際の制約を回避するため、アダプターでは dbt の設定パラメーター incremental_strategy を使用します。これには append を設定できます。この値を設定すると、更新された行はターゲットテーブル (つまり imdb_dbt.actor_summary) に直接 insert され、一時テーブルは作成されません。注: append only モードを使うには、データが不変であるか、重複を許容できる必要があります。変更された行に対応する増分テーブルモデルが必要な場合は、このモードは使用しないでください。

このモードを説明するために、新しい俳優をもう 1 人追加し、incremental_strategy='append' を指定して dbt run を再実行します。

actor_summary.sql で append only モードを設定します:

{{ config(order_by='(updated_at, id, name)', engine='MergeTree()', materialized='incremental', unique_key='id', incremental_strategy='append') }}

もう 1 人の有名な俳優、Danny DeBito を追加します

INSERT INTO imdb.actors VALUES (845467, 'Danny', 'DeBito', 'M');

Danny をランダムに選んだ 920 本の映画に出演させます。

INSERT INTO imdb.roles
SELECT now() as created_at, 845467 as actor_id, id as movie_id, 'Himself' as role
FROM imdb.movies
LIMIT 920 OFFSET 10000;

dbt run を実行し、Danny が actor_summary テーブルに追加されたことを確認します

clickhouse-user@clickhouse:~/imdb$ dbt run
16:12:16  Running with dbt=1.1.0
16:12:16  Found 1 model, 0 tests, 1 snapshot, 0 analyses, 186 macros, 0 operations, 0 seed files, 6 sources, 0 exposures, 0 metrics
16:12:16
16:12:17  Concurrency: 1 threads (target='dev')
16:12:17
16:12:17  1 of 1 START incremental model imdb_dbt.actor_summary........................... [RUN]
16:12:24  1 of 1 OK created incremental model imdb_dbt.actor_summary...................... [OK in 0.17s]
16:12:24
16:12:24  Finished running 1 incremental model in 0.19s.
16:12:24
16:12:24  Completed successfully
16:12:24
16:12:24  Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1

SELECT * FROM imdb_dbt.actor_summary ORDER BY num_movies DESC LIMIT 3;

+------+-------------------+----------+------------------+------+---------+-------------------+
|id    |name               |num_movies|avg_rank          |genres|directors|updated_at         |
+------+-------------------+----------+------------------+------+---------+-------------------+
|845467|Danny DeBito       |920       |1.4768987303293204|21    |670      |2022-04-26 16:22:06|
|845466|Clicky McClickHouse|910       |1.4687938697032283|21    |662      |2022-04-26 16:20:36|
|45332 |Mel Blanc          |909       |5.7884792542982515|19    |148      |2022-04-26 16:17:42|
+------+-------------------+----------+------------------+------+---------+-------------------+

この増分処理が "Clicky" の insert と比べて大幅に高速であることに注目してください。

query_log テーブルをもう一度確認すると、2 回の増分実行の違いが分かります:

INSERT INTO imdb_dbt.actor_summary ("id", "name", "num_movies", "avg_rank", "genres", "directors", "updated_at")
WITH actor_summary AS (
   SELECT id,
      any(actor_name) AS name,
      uniqExact(movie_id)    AS num_movies,
      avg(rank)                AS avg_rank,
      uniqExact(genre)         AS genres,
      uniqExact(director_name) AS directors,
      max(created_at) AS updated_at
   FROM (
      SELECT imdb.actors.id AS id,
         concat(imdb.actors.first_name, ' ', imdb.actors.last_name) AS actor_name,
         imdb.movies.id AS movie_id,
         imdb.movies.rank AS rank,
         genre,
         concat(imdb.directors.first_name, ' ', imdb.directors.last_name) AS director_name,
         created_at
      FROM imdb.actors
         JOIN imdb.roles ON imdb.roles.actor_id = imdb.actors.id
         LEFT OUTER JOIN imdb.movies ON imdb.movies.id = imdb.roles.movie_id
         LEFT OUTER JOIN imdb.genres ON imdb.genres.movie_id = imdb.movies.id
         LEFT OUTER JOIN imdb.movie_directors ON imdb.movie_directors.movie_id = imdb.movies.id
         LEFT OUTER JOIN imdb.directors ON imdb.directors.id = imdb.movie_directors.director_id
   )
   GROUP BY id
)

SELECT *
FROM actor_summary
-- このフィルターはインクリメンタル実行時にのみ適用される
WHERE id > (SELECT max(id) FROM imdb_dbt.actor_summary) OR updated_at > (SELECT max(updated_at) FROM imdb_dbt.actor_summary)

この実行では、新しい行だけが直接 imdb_dbt.actor_summary テーブルに追加され、テーブルは作成されません。

削除と挿入モード (実験的)

これまでの ClickHouse では、更新と削除のサポートは、非同期の Mutations による限定的なものにとどまっていました。これは非常に IO 負荷が高くなる可能性があるため、通常は避けるべきです。

ClickHouse 22.8 では論理削除、ClickHouse 25.7 では論理更新が導入されました。これらの機能により、単一の更新クエリによる変更は、非同期で実体化される場合でも、ユーザーからは即座に反映されたように見えます。

このモードは、たとえば incremental_strategy パラメータでモデルに設定できます。

{{ config(order_by='(updated_at, id, name)', engine='MergeTree()', materialized='incremental', unique_key='id', incremental_strategy='delete+insert') }}

この戦略は対象の model のテーブルを直接操作するため、処理中に問題が発生すると、インクリメンタル model 内のデータが無効な状態になる可能性があります。つまり、atomic な更新ではありません。

要約すると、このアプローチでは次の処理を行います。

アダプターが一時テーブル actor_sumary__dbt_tmp を作成します。変更された行はこのテーブルに書き込まれます。
現在の actor_summary テーブルに対して DELETE を実行します。actor_sumary__dbt_tmp にある id をもとに行が削除されます。
actor_sumary__dbt_tmp の行を、INSERT INTO actor_summary SELECT * FROM actor_sumary__dbt_tmp を使って actor_summary に挿入します。

このプロセスを以下に示します。

`insert_overwrite` モード (実験的)

次の手順を実行します。

インクリメンタルモデルのリレーションと同じ構造を持つステージング (一時) テーブルを作成します: CREATE TABLE {staging} AS {target}。
新しいレコード (SELECT によって生成されたもの) のみをステージングテーブルに挿入します。
新しいパーティション (ステージングテーブルに存在するもの) のみをターゲットテーブル内で置き換えます。

このアプローチには、次の利点があります。

テーブル全体をコピーしないため、デフォルトの戦略より高速です。
INSERT 操作が正常に完了するまで元のテーブルを変更しないため、他の戦略より安全です。途中で障害が発生した場合でも、元のテーブルは変更されません。
データエンジニアリングにおける「パーティション不変性」というベストプラクティスに沿っています。これにより、インクリメンタル処理や並列データ処理、ロールバックなどが簡単になります。

スナップショットの作成

dbtのスナップショットを使用すると、変更可能なモデルに対する変更履歴を時系列で記録できます。これにより、モデルに対して特定時点のクエリを実行でき、アナリストはモデルの過去の状態を「時点をさかのぼって」確認できます。これは、行がいつ有効だったかを開始日と終了日のカラムに記録する type-2 Slowly Changing Dimensions を使用して実現されます。この機能は ClickHouse アダプターでサポートされており、以下で説明します。

この例では、Creating an Incremental Table Model を完了していることを前提としています。actor_summary.sql で inserts_only=True を設定していないことを確認してください。models/actor_summary.sql は次のようになっているはずです:

   {{ config(order_by='(updated_at, id, name)', engine='MergeTree()', materialized='incremental', unique_key='id') }}

   with actor_summary as (
       SELECT id,
           any(actor_name) as name,
           uniqExact(movie_id)    as num_movies,
           avg(rank)                as avg_rank,
           uniqExact(genre)         as genres,
           uniqExact(director_name) as directors,
           max(created_at) as updated_at
       FROM (
           SELECT {{ source('imdb', 'actors') }}.id as id,
               concat({{ source('imdb', 'actors') }}.first_name, ' ', {{ source('imdb', 'actors') }}.last_name) as actor_name,
               {{ source('imdb', 'movies') }}.id as movie_id,
               {{ source('imdb', 'movies') }}.rank as rank,
               genre,
               concat({{ source('imdb', 'directors') }}.first_name, ' ', {{ source('imdb', 'directors') }}.last_name) as director_name,
               created_at
       FROM {{ source('imdb', 'actors') }}
           JOIN {{ source('imdb', 'roles') }} ON {{ source('imdb', 'roles') }}.actor_id = {{ source('imdb', 'actors') }}.id
           LEFT OUTER JOIN {{ source('imdb', 'movies') }} ON {{ source('imdb', 'movies') }}.id = {{ source('imdb', 'roles') }}.movie_id
           LEFT OUTER JOIN {{ source('imdb', 'genres') }} ON {{ source('imdb', 'genres') }}.movie_id = {{ source('imdb', 'movies') }}.id
           LEFT OUTER JOIN {{ source('imdb', 'movie_directors') }} ON {{ source('imdb', 'movie_directors') }}.movie_id = {{ source('imdb', 'movies') }}.id
           LEFT OUTER JOIN {{ source('imdb', 'directors') }} ON {{ source('imdb', 'directors') }}.id = {{ source('imdb', 'movie_directors') }}.director_id
       )
       GROUP BY id
   )
   select *
   from actor_summary

   {% if is_incremental() %}

   -- このフィルターはインクリメンタル実行時にのみ適用されます
   where id > (select max(id) from {{ this }}) or updated_at > (select max(updated_at) from {{this}})

   {% endif %}

snapshots ディレクトリに actor_summary ファイルを作成します。
```
 touch snapshots/actor_summary.sql
```

actor_summary.sql ファイルの内容を、以下のように更新します。

{% snapshot actor_summary_snapshot %}

{{
config(
target_schema='snapshots',
unique_key='id',
strategy='timestamp',
updated_at='updated_at',
)
}}

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

{% endsnapshot %}

この内容について、いくつか補足します。

select クエリは、時間の経過に沿ってスナップショットとして保持したい結果を定義します。ref 関数は、先ほど作成した actor_summary モデルを参照するために使用します。
レコードの変更を示すために、timestamp カラムが必要です。ここでは updated_at カラム (インクリメンタルテーブルモデルの作成を参照) を使用できます。strategy パラメータは、更新の判定に timestamp を使用することを示し、updated_at パラメータは使用するカラムを指定します。これがモデルに存在しない場合は、代わりに check strategy を使用することもできます。これはかなり非効率で、比較対象とするカラムの一覧をユーザーが指定する必要があります。dbt はこれらのカラムの現在値と履歴値を比較し、変更があれば記録します (同一であれば何もしません) 。

dbt snapshot コマンドを実行します。

clickhouse-user@clickhouse:~/imdb$ dbt snapshot
13:26:23  Running with dbt=1.1.0
13:26:23  Found 1 model, 0 tests, 1 snapshot, 0 analyses, 181 macros, 0 operations, 0 seed files, 3 sources, 0 exposures, 0 metrics
13:26:23
13:26:25  Concurrency: 1 threads (target='dev')
13:26:25
13:26:25  1 of 1 START snapshot snapshots.actor_summary_snapshot...................... [RUN]
13:26:25  1 of 1 OK snapshotted snapshots.actor_summary_snapshot...................... [OK in 0.79s]
13:26:25
13:26:25  Finished running 1 snapshot in 2.11s.
13:26:25
13:26:25  Completed successfully
13:26:25
13:26:25  Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1

actor_summary_snapshot テーブルが snapshots DB に作成されていることに注目してください (これは target_schema パラメータによって決まります) 。

このデータをサンプリングすると、dbt によって dbt_valid_from と dbt_valid_to というカラムが追加されていることがわかります。後者の値は null になっています。以降の実行でこれが更新されます。

SELECT id, name, num_movies, dbt_valid_from, dbt_valid_to FROM snapshots.actor_summary_snapshot ORDER BY num_movies DESC LIMIT 5;

+------+----------+------------+----------+-------------------+------------+
|id    |first_name|last_name   |num_movies|dbt_valid_from     |dbt_valid_to|
+------+----------+------------+----------+-------------------+------------+
|845467|Danny     |DeBito      |920       |2022-05-25 19:33:32|NULL        |
|845466|Clicky    |McClickHouse|910       |2022-05-25 19:32:34|NULL        |
|45332 |Mel       |Blanc       |909       |2022-05-25 19:31:47|NULL        |
|621468|Bess      |Flowers     |672       |2022-05-25 19:31:47|NULL        |
|283127|Tom       |London      |549       |2022-05-25 19:31:47|NULL        |
+------+----------+------------+----------+-------------------+------------+

お気に入りの俳優、Clicky McClickHouseをさらに10本の映画に出演させましょう。

INSERT INTO imdb.roles
SELECT now() as created_at, 845466 as actor_id, rand(number) % 412320 as movie_id, 'Himself' as role
FROM system.numbers
LIMIT 10;

imdb ディレクトリで dbt run コマンドを再実行します。これにより、インクリメンタルモデルが更新されます。完了したら、変更を記録するために dbt snapshot を実行します。

clickhouse-user@clickhouse:~/imdb$ dbt run
13:46:14  Running with dbt=1.1.0
13:46:14  Found 1 model, 0 tests, 1 snapshot, 0 analyses, 181 macros, 0 operations, 0 seed files, 3 sources, 0 exposures, 0 metrics
13:46:14
13:46:15  Concurrency: 1 threads (target='dev')
13:46:15
13:46:15  1 of 1 START incremental model imdb_dbt.actor_summary....................... [RUN]
13:46:18  1 of 1 OK created incremental model imdb_dbt.actor_summary.................. [OK in 2.76s]
13:46:18
13:46:18  Finished running 1 incremental model in 3.73s.
13:46:18
13:46:18  Completed successfully
13:46:18
13:46:18  Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1

clickhouse-user@clickhouse:~/imdb$ dbt snapshot
13:46:26  Running with dbt=1.1.0
13:46:26  Found 1 model, 0 tests, 1 snapshot, 0 analyses, 181 macros, 0 operations, 0 seed files, 3 sources, 0 exposures, 0 metrics
13:46:26
13:46:27  Concurrency: 1 threads (target='dev')
13:46:27
13:46:27  1 of 1 START snapshot snapshots.actor_summary_snapshot...................... [RUN]
13:46:31  1 of 1 OK snapshotted snapshots.actor_summary_snapshot...................... [OK in 4.05s]
13:46:31
13:46:31  Finished running 1 snapshot in 5.02s.
13:46:31
13:46:31  Completed successfully
13:46:31
13:46:31  Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1

ここでスナップショットをクエリすると、Clicky McClickHouse に対応する行が 2 行あることがわかります。前のエントリには dbt_valid_to の値が入るようになりました。新しい値は、dbt_valid_from カラムに同じ値で記録され、dbt_valid_to の値は null になっています。新しい行があった場合は、それらもスナップショットに追加されます。

SELECT id, name, num_movies, dbt_valid_from, dbt_valid_to FROM snapshots.actor_summary_snapshot ORDER BY num_movies DESC LIMIT 5;

+------+----------+------------+----------+-------------------+-------------------+
|id    |first_name|last_name   |num_movies|dbt_valid_from     |dbt_valid_to       |
+------+----------+------------+----------+-------------------+-------------------+
|845467|Danny     |DeBito      |920       |2022-05-25 19:33:32|NULL               |
|845466|Clicky    |McClickHouse|920       |2022-05-25 19:34:37|NULL               |
|845466|Clicky    |McClickHouse|910       |2022-05-25 19:32:34|2022-05-25 19:34:37|
|45332 |Mel       |Blanc       |909       |2022-05-25 19:31:47|NULL               |
|621468|Bess      |Flowers     |672       |2022-05-25 19:31:47|NULL               |
+------+----------+------------+----------+-------------------+-------------------+

dbtのスナップショットの詳細については、こちらをご覧ください。

seed の使用

dbt には、CSVファイルからデータを読み込む機能があります。この機能は、database の大規模なエクスポートを読み込む用途には適しておらず、主にコード表やディクショナリなどで使われる小さなファイル向けに設計されています。たとえば、国コードを国名に対応付けるようなケースです。簡単な例として、seed 機能を使ってジャンルコードの一覧を生成し、アップロードします。

既存のデータセットからジャンルコードの一覧を生成します。dbt ディレクトリで、clickhouse-client を使用して seeds/genre_codes.csv ファイルを作成します。

clickhouse-user@clickhouse:~/imdb$ clickhouse-client --password <password> --query
"SELECT genre, ucase(substring(genre, 1, 3)) as code FROM imdb.genres GROUP BY genre
LIMIT 100 FORMAT CSVWithNames" > seeds/genre_codes.csv

dbt seed コマンドを実行します。これにより、CSV ファイルの行を使って、database imdb_dbt に新しい table genre_codes が作成されます (スキーマ設定で定義したとおりです) 。

clickhouse-user@clickhouse:~/imdb$ dbt seed
17:03:23  Running with dbt=1.1.0
17:03:23  Found 1 model, 0 tests, 1 snapshot, 0 analyses, 181 macros, 0 operations, 1 seed file, 6 sources, 0 exposures, 0 metrics
17:03:23
17:03:24  Concurrency: 1 threads (target='dev')
17:03:24
17:03:24  1 of 1 START seed file imdb_dbt.genre_codes..................................... [RUN]
17:03:24  1 of 1 OK loaded seed file imdb_dbt.genre_codes................................. [INSERT 21 in 0.65s]
17:03:24
17:03:24  Finished running 1 seed in 1.62s.
17:03:24
17:03:24  Completed successfully
17:03:24
17:03:24  Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1

これらが読み込まれたことを確認します。

SELECT * FROM imdb_dbt.genre_codes LIMIT 10;

+-------+----+
|genre  |code|
+-------+----+
|Drama  |DRA |
|Romance|ROM |
|Short  |SHO |
|Mystery|MYS |
|Adult  |ADU |
|Family |FAM |

|Action |ACT |
|Sci-Fi |SCI |
|Horror |HOR |
|War    |WAR |
+-------+----+=

追加情報

これまでのガイドで紹介しているのは、dbt の機能のほんの一部にすぎません。詳しくは、充実した dbt documentation をご覧ください。