findscripter 何时使用
适用：
- 用 dbt 在数据仓库上搭建数据转换管道（ELT）。
- 把模型按 staging / intermediate / marts 分层组织（Medallion 风格）。
- 为模型加数据质量测试、列/模型文档、source 新鲜度检查。
- 为大表（经验阈值 > 100 万行）构建增量模型。
- 初始化或规范 dbt 项目结构与命名约定。
不该用（负边界）：
- 项目未使用 dbt，也没有仓库支撑的工作流。
- 只需写一次性的即席 SQL 查询，不需要建模与版本化。
- 没有源数据或 schema 的访问权限。
步骤
- 定模型分层、命名与归属：
sources/ → staging/ → intermediate/ → marts/。 - 在
dbt_project.yml中按层配置物化方式（staging=view、intermediate=ephemeral、marts=table）。 - 写 source 定义（含
freshness）与 staging 模型（1:1 对应源、轻清洗、重命名）。 - 在 intermediate 写业务逻辑（join/聚合），在 marts 产出
dim_/fct_。 - 用
schema.yml加测试（unique/not_null/relationships/accepted_values）与列文档。 - 大表选增量物化与增量策略；用 selector 与
dbt build跑 CI。
指令
- 先建 staging 层一次性清洗，全局复用；禁止 raw 直连 mart（技术债）。
- 测试要狠：主键 unique+not_null、外键 relationships、枚举 accepted_values。
- 一切皆文档：列描述、模型描述齐全。
- 增量优先用于大表；late-arriving 数据用
merge策略。 - 不硬编码日期，改用
{{ var('start_date') }}；重复逻辑抽成 macro。 - 监控 source 新鲜度，不在生产 target 上测试。
- 命名前缀：Staging=
stg_（如stg_stripe__payments），Intermediate=int_，Marts=dim_/fct_。 - 需要更完整的代码样例时，参见源仓库
resources/implementation-playbook.md。
示例
项目配置（dbt_project.yml）按层设置默认物化：
models:
analytics:
staging:
+materialized: view
+schema: staging
intermediate:
+materialized: ephemeral
marts:
+materialized: table
+schema: analytics
Source 定义 + 新鲜度 + 测试（_stripe__sources.yml）：
version: 2
sources:
- name: stripe
database: raw
schema: stripe
loaded_at_field: _fivetran_synced
freshness:
warn_after: {count: 12, period: hour}
error_after: {count: 24, period: hour}
tables:
- name: payments
columns:
- name: id
tests: [unique, not_null]
- name: customer_id
tests:
- not_null
- relationships:
to: source('stripe', 'customers')
field: id
增量 staging 模型（stg_stripe__payments.sql）：
{{ config(materialized='incremental', unique_key='payment_id',
}}
with source as (
select * from {{ source('stripe', 'payments') }}
{% if is_incremental() %}
where _fivetran_synced > (select max(_loaded_at) from {{ this }})
{% endif %}
)
select
id as payment_id,
customer_id,
amount / 100.0 as amount, -- 分转元
status as payment_status,
created as created_at,
_fivetran_synced as _loaded_at
from source
Mart 维度表，含代理键与计算字段（dim_customers.sql 节选）：
{{ config(materialized='table', unique_key='customer_id') }}
select
{{ dbt_utils.generate_surrogate_key(['customer_id']) }} as customer_key,
customer_id,
case
when lifetime_value >= 1000 then 'high'
when lifetime_value >= 100 then 'medium'
else 'low'
end as customer_tier
from {{ ref('int_payments_pivoted_to_customer') }}
模型测试 + 文档（_core__models.yml 节选）：
version: 2
models:
- name: dim_customers
columns:
- name: customer_key
tests: [unique, not_null]
- name: customer_tier
tests:
- accepted_values:
values: ['high', 'medium', 'low']
复用 macro（DRY）：
{% macro cents_to_dollars(column_name, precision=2) %}
round({{ column_name }} / 100.0, {{ precision }})
{% endmacro %}
{% macro limit_data_in_dev(column_name, days=3) %}
{% if target.name == 'dev' %}
where {{ column_name }} >= dateadd(day, -{{ days }}, current_date)
{% endif %}
{% endmacro %}
增量策略选型：
-- merge：适合 late-arriving 数据
{{ config(materialized='incremental', unique_key='id',
incremental_strategy='merge',
merge_update_columns=['status','amount','updated_at']) }}
-- insert_overwrite：按分区覆盖
{{ config(materialized='incremental',
incremental_strategy='insert_overwrite',
partition_by={"field":"created_date","data_type":"date","granularity":"day"}) }}
常用 dbt 命令：
dbt run --select staging # 只跑 staging 层
dbt run --select +fct_orders # fct_orders 及其上游
dbt run --select fct_orders+ # fct_orders 及其下游
dbt run --full-refresh # 重建增量模型
dbt test --select stg_stripe # 测指定模型
dbt build # 按 DAG 顺序 run + test
dbt docs generate && dbt docs serve
dbt compile # 只编译不执行
dbt ls --select tag:critical # 按 tag 列出模型
注意事项
- 增量模型的
is_incremental()过滤条件必须基于可靠的水位字段（如_loaded_at/updated_at），否则会漏数或重复。 merge策略要求设置unique_key；不同仓库默认增量策略不同（多数为delete+insert）。on_schema_change决定增量时如何处理列变更，源 schema 变动时需复核。- 产出不能替代环境内的真实校验、测试与专家评审；缺少输入、权限、安全边界或验收标准时先停下来澄清。
- 仅在任务明确落在上述范围内时使用本技能。
互见
- 源 Playbook：
resources/implementation-playbook.md（更完整的模式与样例）。 - 官方文档：dbt Docs（docs.getdbt.com）、dbt Best Practices、dbt-utils 包。
采编自 sickn33/antigravity-awesome-skills（MIT 许可）。