指南

本节提供了有关设置 dbt 和 ClickHouse 适配器的指南，并通过一个公开可用的 IMDB 数据集示例说明如何将 dbt 与 ClickHouse 配合使用。该示例涵盖以下步骤：

1. 创建 dbt 项目并设置 ClickHouse 适配器。
2. 定义模型。
3. 更新模型。
4. 创建增量模型。
5. 创建快照模型。
6. 使用 materialized view。

这些指南应结合其余文档、功能和配置以及物化类型参考一并使用。

请按照 dbt 和 ClickHouse 适配器 的设置 部分中的说明准备环境。

重要提示：以下内容已在 Python 3.9 下测试。

dbt 在对高度关系型数据进行建模时表现出色。为便于说明，我们提供了一个小型 IMDB 数据集，其关系型 schema 如下所示。该数据集来自关系型数据集 repository。相较于 dbt 中常见的 schema，这个数据集非常简单，但作为一个易于处理的样本很合适：

如图所示，我们使用其中部分表。

创建以下表：

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|
+------+------------+----------+------------------+-------------+--------------+-------------------+

在后续指南中，我们会将此查询转换为一个模型——并在 ClickHouse 中将其物化为 dbt 视图和表。

1. 创建一个 dbt 项目。在本例中，我们以 imdb source 为项目命名。出现提示时，选择 clickhouse 作为数据库 source。

   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

2. 使用 cd 进入项目目录：

   cd imdb

3. 此时，你需要使用自己选择的文本编辑器。在下面的示例中，我们使用常见的 VS Code。打开 IMDB 目录后，你应该会看到一组 yml 和 sql 文件：

   

4. 更新你的 dbt_project.yml 文件，指定第一个模型 actor_summary，并将 profile 设为 clickhouse_imdb。

   

   

5. 接下来，我们需要向 dbt 提供 ClickHouse 实例的 connection details。将以下内容添加到 ~/.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。有关其他可用设置的说明，请参见这里。

6. 在 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 语句重建为视图。这样无需额外存储数据，但查询速度会比表物化类型慢。

1. 在 imdb 文件夹下，删除目录 models/example：

   clickhouse-user@clickhouse:~/imdb$ rm -rf models/example

2. 在 models 文件夹中的 actors 目录下创建一个新文件。这里创建的每个文件都对应一个 actor 模型：

   clickhouse-user@clickhouse:~/imdb$ mkdir models/actors

3. 在 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 定义了我们的表。之后，这些表就可以在 macro 中使用。编辑 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 中 materialize 为视图。我们的表是通过 schema.yml 文件中的 source 函数引用的，例如 source('imdb', 'movies') 指向 imdb database 中的 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 列。后续会将其用于增量物化。

4. 在 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

5. dbt 会按要求将该模型在 ClickHouse 中表示为一个视图。现在，我们可以直接查询该视图。该视图会创建在 imdb_dbt database 中——这是由 clickhouse_imdb profile 下 ~/.dbt/profiles.yml 文件中的 schema parameter 决定的。

   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。

1. 修改文件 actors_summary.sql，将 materialized 参数设置为 table。注意 ORDER BY 的定义方式，以及这里使用的是 MergeTree 表引擎：

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

2. 在 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

3. 确认表 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)
   +----------------------------------------

4. 确认该表返回的结果与之前的结果一致。注意，现在模型已物化为表，响应时间有了明显改善：

   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。

1. 首先，我们将模型改为 incremental 类型。此更改需要：

   1. unique_key - 为确保适配器能够唯一标识各行，我们必须提供一个 unique_key——在本例中，查询中的 id 字段就足够了。这样可以确保物化后的表中不会出现重复行。有关唯一性约束的更多信息，请参见这里。
   2. Incremental filter - 我们还需要告诉 dbt，在增量运行时应如何识别哪些行发生了变化。这可以通过提供一个增量表达式来实现。对于事件数据，这通常会涉及一个 timestamp；因此这里使用的是 updated_at timestamp 字段。该列在插入行时默认值为 now()，从而可以识别新增的角色。此外，我们还需要识别另一种情况，即新增了 actor。使用 {{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 表中的更新和新增数据。若要覆盖所有表，建议将此模型拆分为多个子模型——每个子模型都有各自的增量条件。随后，这些模型可以相互引用并关联起来。有关模型间交叉引用的更多信息，请参见此处。

2. 执行 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|
   +------+------------+----------+------------------+------+---------+-------------------+

3. 现在，我们将向模型添加数据，以演示增量更新。将我们的演员 "Clicky McClickHouse" 添加到 actors 表中：

   INSERT INTO imdb.actors VALUES (845466, 'Clicky', 'McClickHouse', 'M');

4. 让“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;

5. 通过查询底层源表并绕过所有 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|
   +------+-------------------+----------+------------------+------+---------+-------------------+

6. 运行一次 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;

将上述查询调整为实际执行的时间范围。结果如何验证留给用户自行检查，这里重点说明 适配器 执行增量更新时采用的一般策略：

1. 适配器 会创建一个临时表 actor_sumary__dbt_tmp。发生变化的行会被流式写入该表。
2. 接着会创建一个新表 actor_summary_new,。随后，旧表中的行会从旧表流式传输到新表，同时检查这些行的 ID 是否不存在于临时表中。这样可以有效处理更新和重复数据。
3. 临时表中的结果会被流式传输到新的 actor_summary 表中：
4. 最后，通过 EXCHANGE TABLES 语句以原子方式将新表与旧版本交换。随后再删除旧表和临时表。

如下图所示：

这种策略在非常大的模型上可能会遇到一些挑战。更多细节请参见 限制。

为克服增量模型处理大型数据集时的局限性，适配器 使用 dbt 配置参数 incremental_strategy。可将其设置为 append。设置后，更新的行会直接插入目标表 (即 imdb_dbt.actor_summary) ，不会创建临时表。 注意：仅追加模式要求数据是不可变的，或者可以接受重复数据。如果你需要支持已修改行的增量表模型，请不要使用此模式！

为了演示此模式，我们将再添加一位新演员，并在 incremental_strategy='append' 的情况下重新执行 dbt run。

1. 在 actor_summary.sql 中配置仅追加模式：

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

2. 再添加一位著名演员 —— Danny DeBito

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

3. 让 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;

4. 执行一次 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”时相比，这次增量运行快了很多。

再次检查 query_log 表，可以看出这两次增量运行之间的差异：

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 对更新和删除的支持都比较有限，主要通过异步的变更实现。这类操作可能会产生极高的 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') }}

该策略直接对目标模型的表进行操作，因此如果在操作过程中出现问题，增量模型中的数据很可能会处于无效状态——因为这里没有原子更新。

总结来说，这种方法会：

1. 适配器会创建一个临时表 actor_sumary__dbt_tmp。发生变更的行会被流式写入该表。
2. 对当前的 actor_summary 表执行一条 DELETE。根据 actor_sumary__dbt_tmp 中的 id 删除对应的行。
3. 使用 INSERT INTO actor_summary SELECT * FROM actor_sumary__dbt_tmp 将 actor_sumary__dbt_tmp 中的行插入 actor_summary。

该过程如下所示：

执行以下步骤：

1. 创建一个与增量模型 relation 结构相同的暂存 (临时) 表：CREATE TABLE {staging} AS {target}。
2. 仅将新记录 (由 SELECT 生成) 插入暂存表。
3. 仅将新分区 (即暂存表中存在的分区) 替换到目标表中。

这种方法有以下优点：

• 它比默认策略更快，因为无需复制整个表。
• 它比其他策略更安全，因为在 INSERT 操作成功完成之前，不会修改原始表：如果中途失败，原始表不会被修改。
• 它实现了数据工程中“分区不可变性”的最佳实践，从而简化增量和并行数据处理、回滚等操作。

dbt 快照可用于记录可变模型随时间发生的变化。这样一来，就能对模型执行时间点查询，使分析人员能够"回溯"查看模型先前的状态。这是通过使用 type-2 Slowly Changing Dimensions 实现的，其中起始日期列和结束日期列用于记录某一行在何时有效。ClickHouse 适配器 支持此功能，下面将进行演示。

本示例假定你已经完成了创建增量表模型。请确保你的 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 %}

1. 在 snapshots 目录中创建一个 actor_summary 文件。

    touch snapshots/actor_summary.sql

2. 将 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 模型。
• 我们需要一个时间戳列来标识记录变更。这里可以使用 updated_at 列 (参见创建增量表模型) 。strategy 参数表示我们使用时间戳来标记更新，而 updated_at 参数则指定使用哪一列。如果你的模型中没有这个列，也可以改用 check 策略。这种方式效率会低很多，并且需要用户指定要比较的列列表。dbt 会比较这些列的当前值和历史值，并记录所有变化 (如果值相同，则不执行任何操作) 。

3. 运行命令 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

请注意，snapshots DB 中已创建名为 actor_summary_snapshot 的表 (由 target_schema parameter 决定) 。

4. 对这些数据进行抽样后，你会看到 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        |
   +------+----------+------------+----------+-------------------+------------+

5. 让我们最喜欢的演员 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;

6. 在 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

7. 如果我们现在查询这个快照，会发现 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 快照的更多信息，请参见此处。

dbt 支持从 CSV 文件加载数据。不过，这一功能并不适合加载数据库的大型导出数据，更适用于通常作为代码表和字典的小型文件，例如将国家代码映射为国家名称。下面通过一个简单示例，使用 seed 功能生成并上传一份类型代码列表。

1. 我们先从现有数据集中生成一份类型代码列表。在 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

2. 执行 dbt seed 命令。这会在数据库 imdb_dbt 中创建一个新表 genre_codes (由 schema 配置定义) ，并将 csv 文件中的行加载到该表中。

   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

3. 确认这些数据已加载：

   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 文档。