mz-dbt

dbt-materialize

dbt-materialize 适配器可让你使用 dbt 基于模型的工作流管理 Materialize 流水线。它扩展了 dbt-postgres，添加了 Materialize 专用的物化方式、蓝绿部署宏和集群管理功能。

如何使用此技能

此技能是自包含的。帮助用户编写 dbt 模型、配置配置文件、运行部署和排查问题所需的所有信息都已记录在下文中。无需访问 dbt-materialize 源代码或 Materialize monorepo。

1. 用户想要设置 dbt 与 Materialize：参见下文的配置文件配置。
2. 用户询问特定的物化方式：参见物化方式一节。
3. 用户想要进行蓝绿部署：参见蓝绿部署一节。
4. 用户询问索引、集群或配置：参见索引、严格模式或集群管理一节。
5. 用户遇到错误或意外行为：参见文末的注意事项。

如需深入了解适配器内部机制（很少需要），源代码位于 GitHub 上 MaterializeInc/materialize 仓库的 misc/dbt-materialize/ 目录中。面向用户的官方文档位于 https://materialize.com/docs/manage/dbt/。

安装

pip install dbt-core dbt-materialize

需要 Materialize v0.68.0+。

配置文件配置

在 ~/.dbt/profiles.yml 中：

my_project:
  target: dev
  outputs:
    dev:
      type: materialize
      host: <host>
      port: 6875
      user: <user>
      pass: <password>
      dbname: materialize
      schema: <dbt_schema>
      cluster: <cluster_name>       # optional, defaults to user's default cluster
      sslmode: require
      threads: 1                    # must be 1, Materialize does not support concurrent sessions

要点：

• threads 必须设为 1。Materialize 不支持并发 DDL 会话。
• cluster 是可选项，但建议设置。若省略，则使用用户的默认集群。
• 始终启用自动提交。Materialize 不支持在事务中执行任意查询。
• 生产部署请使用服务账号。

物化方式

materialized_view（主要模式）

流式转换中最常用的物化方式。创建增量维护的视图，并在上游数据发生变化时自动更新。

-- models/order_totals.sql
{{ config(materialized='materialized_view') }}

SELECT
    customer_id,
    COUNT(*) AS order_count,
    SUM(amount) AS total_amount
FROM {{ ref('orders') }}
GROUP BY customer_id

可通过配置使用 WITH 子句选项：

config:
  materialized: materialized_view
  cluster: analytics
  refresh_interval:
    every: '1 hour'
    aligned_to: '00:00:00'
  partition_by: [region, created_date]
  retain_history: '1 hour'

配置	用途
cluster	在指定集群上运行
refresh_interval	使用基于计划的刷新，而非持续刷新
partition_by	用于筛选下推的内部存储排序
retain_history	在一段时间内保留历史数据

view

标准 SQL 视图。适用于无需将结果存储在内存中的轻量级转换。

{{ config(materialized='view') }}
SELECT * FROM {{ ref('raw_events') }} WHERE status = 'active'

source

定义外部数据源（Kafka、PostgreSQL、MySQL）。需要预先创建 CONNECTION，且该操作需在 dbt 之外完成（例如通过 Terraform 或 SQL）。

-- models/staging/kafka_events.sql
{{ config(materialized='source') }}
FROM KAFKA CONNECTION kafka_connection (TOPIC 'events')
FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection

-- models/staging/pg_source.sql
{{ config(materialized='source') }}
FROM POSTGRES CONNECTION pg_connection (PUBLICATION 'mz_source')
FOR ALL TABLES

使用 FOR ALL TABLES 的数据源会自动创建子数据源。在 YAML 文件中将这些子数据源定义为 dbt 数据源，以便在下游模型中引用。

当模型依赖数据源的子数据源时，使用 -- depends_on: {{ ref('pg_source') }} 强制按顺序执行。

source_table

较新的物化方式，用于从数据源创建表，并更好地支持数据源版本控制。

{{ config(materialized='source_table', strict_mode=True) }}
FROM SOURCE {{ ref('kafka_events') }} (REFERENCE "events_topic")

设置 strict_mode: True 后，如果对象已存在，则跳过重新创建。使用 force_recreate: True 可覆盖此设置。

sink

将数据导出到外部系统。需要预先创建 CONNECTION。

{{ config(materialized='sink') }}
FROM {{ ref('order_totals') }}
INTO KAFKA CONNECTION kafka_connection (TOPIC 'order-totals-output')
FORMAT JSON
ENVELOPE DEBEZIUM

sink 必须位于专用架构中（启用 strict_mode 时会强制执行）。蓝绿部署期间，sink 会自动切换到新的上游定义。

table

在 Materialize 中映射为 CREATE MATERIALIZED VIEW（计划支持真正的表）。为清晰起见，请直接使用 materialized_view。

seed

标准 dbt seed 可正常使用。它会创建表并插入 CSV 数据。为 seed 操作配置集群：

seeds:
  +cluster: seed_cluster

不支持的物化方式

• incremental：请改用 materialized_view。Materialize 原生处理增量更新。
• snapshot：不支持。可考虑使用带 retain_history 的 materialized_view。

索引

为视图、物化视图、数据源和表添加索引，以支持快速点查：

config:
  indexes:
    - columns: [customer_id]
    - columns: [region, created_date]
      cluster: serving_cluster
    - default: true              # index all columns

索引会在关系创建后创建。如果未指定 cluster，则使用模型的集群。

严格模式

在模型配置中启用 strict_mode: True，以强制执行生产环境隔离规则：

• 源表只能与其他源表和种子共享架构
• Sink 只能与其他 Sink 共享架构
• 源只能与其他源共享架构
• 标准模型（视图、物化视图）不能与源、Sink 或源表共享架构

验证会在编译时使用 dbt 清单执行。

蓝绿部署

通过构建到并行环境并原子切换，实现零停机部署。

前置条件

在 dbt_project.yml 中配置部署目标：

vars:
  deployment:
    default:
      clusters:
        - production_cluster
      schemas:
        - public

运行 dbt 的角色必须拥有要部署的架构和集群。Sink 必须位于专用架构和集群中，且不得列在部署配置中。

工作流

1. 初始化部署环境：

dbt run-operation deploy_init

通过克隆生产环境中的配置、授权和默认权限，创建 <schema>_dbt_deploy 和 <cluster>_dbt_deploy。验证生产架构和集群是否存在、其中是否不包含 Sink，以及执行角色是否具有所有权。使用 --args '{ignore_existing_objects: True}' 可跳过部署对象已存在时产生的错误。

对于托管集群，会创建大小、复制因子和计划相同的新集群。对于非托管集群，会克隆副本，包括其大小和可用区。

2. 构建到绿色环境：

dbt run --vars 'deploy: True' --exclude config.materialized:source config.materialized:sink

模型会自动路由到 _dbt_deploy 架构和集群。务必排除源和 Sink。

如果遇到 String 'deploy:' is not valid YAML（Windows/PowerShell 中很常见），请改用 --vars "{\"deploy\": true}"。

3. 等待数据填充（推荐）：

dbt run-operation deploy_await

通过单个汇总查询轮询集群就绪状态（降低目录服务端负载）。默认情况下，每 15 秒轮询一次，并等待延迟低于 1 秒。使用 --args '{poll_interval: 30, lag_threshold: "5s"}' 配置。

就绪检查按以下顺序评估三项内容：

• 副本健康状况：标记过去 24 小时内 OOM 终止次数达到 3 次或以上的所有副本所在集群
• 数据填充状态：统计每个集群中已填充与对象总数
• 延迟阈值：检查全局墙钟延迟是否低于阈值

4. 提升到生产环境：

dbt run-operation deploy_promote

在单个事务中执行原子 ALTER SCHEMA ... SWAP WITH 和 ALTER CLUSTER ... SWAP WITH。先切换架构，再切换集群，以避免对象引用失效。遇到并发 DDL 冲突时会自动重试（SQLSTATE 40001，默认最多重试 3 次）。

切换后，宏会找出上游依赖发生变化的 Sink，并运行 ALTER SINK ... SET FROM，将其切换到新的上游定义。它还会为已部署的架构添加注释，其中包含部署用户、时间戳和 git 提交 SHA。

参数	默认值	用途
dry_run	false	打印切换命令，但不执行
wait	false	切换前等待数据填充
poll_interval	15	就绪检查之间的间隔秒数（当 wait 为 true 时）
lag_threshold	"1s"	就绪状态允许的最大延迟（当 wait 为 true 时）
max_retries	3	并发 DDL 冲突时的重试次数

5. 清理：

dbt run-operation deploy_cleanup

删除 _dbt_deploy 架构和集群（CASCADE）。

切换集群上任何正在执行的 SUBSCRIBE 命令都会中断。重试时，客户端会自动重新连接到新部署的集群。

集群管理宏

# Create a managed cluster
dbt run-operation create_cluster --args '{cluster_name: analytics, size: "100cc", replication_factor: 1}'

# Drop a cluster
dbt run-operation drop_cluster --args '{cluster: analytics}'

# Set the active cluster for a session
dbt run-operation set_cluster --args '{cluster: analytics}'

精简部署

适用于可接受停机的开发环境：

1. 每次部署后，将生产环境的 manifest.json 存储到 Blob 存储中
2. 在 CI 中下载该清单
3. 仅运行已修改的模型：

   dbt build --select state:modified+ --state ./ --defer
   

节点选择

# Run specific models
dbt run --select "model_name"
dbt run --select "tag:nightly"

# Exclude sources and sinks (common in blue/green)
dbt run --exclude config.materialized:source config.materialized:sink

也可以使用 selectors.yml：

selectors:
  - name: exclude_sources_and_sinks
    default: true
    definition:
      union:
        - method: fqn
          value: "*"
        - exclude:
            - 'config.materialized:source'
            - 'config.materialized:sink'

测试

标准测试可正常运行。配置失败记录的存储位置：

store_failures: true
store_failures_as: materialized_view  # or 'view' (default: 'table')

单元测试需要 dbt-materialize v1.8.0+，且上游依赖必须存在于数据库中：

dbt test --select test_type:unit

不支持源新鲜度检查（根据定义，Materialize 中的源始终是最新的）。

模型契约

启用契约强制执行，在编译时验证列类型：

models:
  - name: order_totals
    config:
      contract:
        enforced: true
    columns:
      - name: customer_id
        data_type: text
        constraints:
          - type: not_null    # the only enforced constraint

仅强制执行 not_null（通过 ASSERT NOT NULL）。不支持其他所有约束类型（unique、primary_key、foreign_key、check）。

注意事项

• threads 必须为 1：Materialize 不支持并发 DDL 会话。将 threads 设为大于 1 会导致错误。
• 不支持事务：始终启用自动提交。某些 dbt 内部宏的行为可能会有所不同。
• LIMIT 性能影响：dbt show 使用 LIMIT，这会影响 Materialize 的性能。请谨慎使用。
• 连接是外部资源：源和接收器需要预先存在的 CONNECTION（通过 Terraform、SQL 或其他工具创建）。dbt 不管理连接。
• 集群必须预先存在：dbt 不会隐式创建集群。运行模型前，请通过 SQL、Terraform 或 create_cluster 宏创建集群。
• 不支持授权配置：dbt 内置的授权配置会引发错误。请使用 Terraform 或 SQL 实现 RBAC。
• 不支持增量物化：请改用 materialized_view；Materialize 会原生处理增量更新。
• 蓝绿部署需要隔离接收器：接收器必须位于专用架构和集群中，且这些架构和集群不得列在部署配置中。
• 关系名称长度限制：255 个字符（远长于 PostgreSQL 的 63 个字符）。