create-dbt-transformations

创建 dbt 转换

使用此技能生成 SQL 模型、声明数据源、更新 schema.yml，并通过 dbt run / dbt test 验证模型。

如需探索 BigQuery（列出数据集、检查原始表、预览行、查看空值率），请加载 explore-lakehouse 技能。如果未安装该技能，请改用：

bq show --format=prettyjson $REVOS_BQ_DATASET.<table>

警告用户：“未安装 explore-lakehouse 技能 — 将使用 bq show 作为备用方案。安装该技能可进行更丰富的架构探索。”

---

第 1 部分：知识库

分层约定

• gold — 可用于报告或下游使用的业务就绪模型。
• silver — 已清理、去重并统一类型的中间层。最低 SQL 层；通过 {{ source('bronze', '<table>') }} 读取原始数据。
• bronze — 不是 SQL 层。仅包含 dbt/models/bronze/schema.yml，用于将原始表声明为 dbt 数据源。dbt/models/bronze/ 下不得放置 .sql 文件。

如果根据上下文无法确定所属层级，请提问（参见检查点 1）。

数据源（bronze 层）

由数据摄取管道加载到数据仓库中的原始表不是 dbt 模型。应将它们声明为 dbt 数据源，以便 silver 模型通过 {{ source() }} 引用。

数据源在 dbt/models/bronze/schema.yml 的 sources: 块中声明，并使用 schema（BigQuery 数据集）：

sources:
  - name: bronze
    schema: "{{ env_var('REVOS_BQ_DATASET') }}"
    tables:
      - name: hubspot_contacts

在 silver SQL 中引用：

-- dbt/models/silver/silver_hubspot_contacts.sql
SELECT * FROM {{ source('bronze', 'hubspot_contacts') }}

{{ source('bronze', 'hubspot_contacts') }} 会解析为 ${REVOS_BQ_DATASET}.hubspot_contacts — 与原始表所在的数据集相同，因此 silver 可以直接访问，无需经过 bronze SQL 视图。

完整的声明模式请参阅 schema-conventions.md。

物化方式

全局继承自 dbt_project.yml — 除非用户明确要求覆盖，否则不要添加 {{ config(materialized=...) }}。

schema.yml 约定

每层在 dbt/models/<layer>/schema.yml 中共用一个文件。将新模型追加到该文件中；切勿为每个模型单独创建 YAML 文件。完整示例以及复合主键 / dbt-utils 模式请参阅 schema-conventions.md。

解析 BigQuery 物理表

物化表位于：$REVOS_BQ_DATASET.<model_name>

何时使用 {{ ref() }}，何时使用 {{ source() }}：

上下文	使用
dbt SQL → 其他 dbt 模型	{{ ref('<model>') }}
dbt SQL → 原始表（silver 从 bronze 读取）	{{ source('bronze', '<table>') }}

silver 是最低 SQL 层 — {{ source('bronze', ...) }} 仅用于 silver。gold 通过 {{ ref() }} 从 silver 读取数据。dbt/models/bronze/ 中没有 SQL 文件。

引用原始表之前，务必先将其声明为数据源。不要使用不带引用的完整限定名称 — 这会绕过 dbt 的依赖关系图和数据源新鲜度跟踪。

标准 dbt 命令

dbt parse                               # validate syntax (no warehouse)
dbt compile --select <model>            # resolve refs, produce compiled SQL
dbt run --select <model>                # execute against warehouse
dbt test --select <model>               # run tests
dbt build --select <model>              # run + test
dbt build --select path:models/<layer>  # entire layer

---

第 2 部分：工作流 — 创建新的 dbt 转换

执行顺序

对于每项转换（一次处理一项，不要批量处理）：

1. 确定目标层——只能是 silver 或 gold（如果不明确，请参阅检查点 1）。拒绝 bronze SQL 模型（参阅检查点 4）。
2. 确定模型名称。
3. 检查该模型是否已存在（如果存在，请参阅检查点 2）。
4. 收集源数据和转换逻辑。对于桥接模型，应用桥接模板（sql-templates.md）。
5. 如果模型读取原始数据，请确保在 dbt/models/bronze/schema.yml 的 bronze 源下声明了每个原始表；如果缺少，则添加。
6. 生成 dbt/models/<silver|gold>/<model_name>.sql。绝不要在 dbt/models/bronze/ 下生成 .sql 文件。
7. 检测主键（如果不明确，请参阅检查点 3）。
8. 在 dbt/models/<layer>/schema.yml 中添加模型条目，并添加 PK 和 FK 测试。参阅 schema-conventions.md。
9. 运行 dbt run --select <model_name> 并报告结果。
10. 运行 dbt test --select <model_name> 并报告结果。
11. 汇总（参阅最终回复格式）。

如果一个请求包含多项转换：按模型顺序重复步骤 1–11。

---

必须向用户确认的检查点

检查点 1：选择层

如果目标层不明显，请询问：

Which layer should this transformation live in?

- gold: business-ready, exposed for reporting or downstream consumption
- silver: cleaned/intermediate, reads raw via `{{ source('bronze', ...) }}`

(bronze is not a SQL layer — it only holds `schema.yml` source declarations.)

用户明确指定层时，目标层即为明确。

检查点 2：模型已存在冲突

如果 dbt/models/<layer>/<model_name>.sql 已存在：

A model named <model_name> already exists at dbt/models/<layer>/<model_name>.sql.

Options:
- overwrite: replace with new transformation
- edit:      modify existing (describe the change)
- rename:    use a different name

如果在其他层中发现该模型，也要提及。

检查点 3：主键不明确

如果主键检测无法得出明确结果：

I could not unambiguously detect the primary key. Candidates:
- <candidate_1>
- <candidate_2>

Which column(s) should be the primary key?

检查点 4：拒绝 Bronze SQL 模型

如果用户明确要求创建 bronze SQL 模型：

Bronze is not a SQL layer in this project — it only holds source
declarations in `dbt/models/bronze/schema.yml`. Silver reads raw data
directly via `{{ source('bronze', '<raw_table>') }}`.

Would you like to create this as a silver model instead?

除 schema.yml 外，不要在 dbt/models/bronze/ 下生成任何文件。

---

主键检测

按顺序应用；首次得出明确结果后即停止：

1. ROW_NUMBER() OVER (PARTITION BY <cols>) = 1 → 分区列为 PK。
2. 对少量列使用 SELECT DISTINCT → 所有选定列构成复合 PK。
3. 最外层使用 GROUP BY <cols> → 分组列为 PK。
4. 名为 id 的单列 → PK。
5. 名为 <entity>_id 且与模型名称词干匹配的单列 → PK。
6. 桥接命名 <entity_a>_<entity_b> → (<entity_a>_id, <entity_b>_id) 复合 PK。

如果以上都无法得出明确结果 → 检查点 3。

外键检测

如果某列符合 <entity>_id 格式，其中 <entity> ≠ 模型自身实体，且该列不属于 PK，并且按设计不可为空，则该列可作为 FK 候选项。仅添加 not_null 测试（默认不添加 relationships 测试）。

时间戳列传递（金层模型）

每个金层模型必须至少传递一个时间戳列，以便下游立方体使用基于 SQL 的 refresh_key（参阅 create-cubes 技能）。优先级如下：

1. 原始表上的摄取时间列（例如，Airbyte 会写入 _airbyte_extracted_at）——存在时应传递。
2. updated_at / modified_at——适用于支持 CDC 的数据流。
3. created_at——适用于仅插入的事实表。

如果上游源没有上述任何列，请在 SQL 注释中注明：-- no timestamp column available from source。

SQL 文件生成

关于以下内容，请参阅 sql-templates.md：

• 标准银层模型模板（通过 {{ source('bronze', ...) }} 读取原始数据）
• 标准金层模型模板（通过 {{ ref() }} 读取银层数据）
• 包含具体示例的桥接模型（JSON 数组）模板
• 桥接模型命名约定和 SQL 内容规则

schema.yml 更新

完整示例（包括源声明、复合 PK 和 dbt-utils 模式）请参阅 schema-conventions.md。

边缘情况

关于缺少 SQL 细节、缺少上游模型、未声明的源以及运行/测试失败处理，请参阅 edge-cases.md。

---

最终回复格式

Created dbt transformation: <model_name>

Layer:           <silver | gold>
File:            dbt/models/<layer>/<model_name>.sql
Materialization: <inherited: table | overridden: <type>>
Primary key:     <pk_column>  (or composite: <col_1>, <col_2>)
Foreign keys:    <fk_1>, <fk_2>  (or "none detected")
schema.yml:      dbt/models/<layer>/schema.yml (entry added)

Tests:
- not_null on <pk>: added
- unique on <pk>: added | skipped: dbt-utils unavailable
- not_null on <fk>: added

Validation:
- dbt run:  passed | failed
- dbt test: passed | failed

Physical table after run:
`<resolved_dataset>.<model_name>`