如何编写dbt命令选择需全量刷新的Schema变更增量模型

dbt 默认的 state:modified 只追踪模型代码、依赖模型的变更，不会主动监控源表的 Schema 变化，所以直接用它达不到你的需求。下面是几种可行的实现方案：

方案一：结合源新鲜度检查（source freshness）筛选目标模型 #

1. 开启源的Schema校验
   在你的 sources.yml 或 dbt_project.yml 中，给需要监控的源添加 schema_check: true，让dbt检查源表实际列和你定义的源列是否匹配：

   sources:
     - name: raw_source
       schema: raw_db
       tables:
         - name: user_events
           schema_check: true
           freshness:
             warn_after: {count: 24, period: hour}
           loaded_at_field: event_timestamp
   

2. 提取Schema变更的源并关联增量模型
   先通过 dbt source freshness 检测出Schema变更的源，再筛选依赖这些源的增量模型，最后执行全量刷新。可以用shell脚本串联：

   # 导出Schema校验失败的源名称（用jq解析JSON输出更可靠）
   changed_sources=$(dbt source freshness --state ./target/previous_state --output json | jq -r '.[] | select(.schema_check.status == "fail") | .source_name')
   
   # 筛选依赖这些源的增量模型
   target_models=$(dbt ls --select $(echo $changed_sources | sed 's/ / source:/g')+ --resource-type model --config.materialized:incremental --output name)
   
   # 对目标模型执行全量刷新
   if [ -n "$target_models" ]; then
     dbt run --select $target_models --full-refresh
   fi
   

   注意：需要提前安装jq来解析JSON输出，./target/previous_state是你上次运行dbt后保存的状态目录（可以用dbt run --state ./target/previous_state生成）。

方案二：自定义源Schema快照对比 #

如果需要更灵活的检测逻辑，可以自己生成源Schema快照并对比：

1. 编写宏生成源Schema快照
   创建一个宏，查询数据库的INFORMATION_SCHEMA获取源表的列信息，保存为JSON快照：

   {% macro snapshot_source_schemas() %}
     {% set snapshot_data = {} %}
     {% for source in graph.sources.values() %}
       {% set column_query %}
         SELECT column_name, data_type 
         FROM {{ source.database }}.INFORMATION_SCHEMA.COLUMNS
         WHERE table_schema = '{{ source.schema }}' AND table_name = '{{ source.name }}'
       {% endset %}
       {% set columns = run_query(column_query).rows | list %}
       {% do snapshot_data.update({source.unique_id: columns}) %}
     {% endfor %}
     {% do write_json('./target/current_source_schema.json', snapshot_data) %}
   {% endmacro %}
   

   把这个宏加到dbt_project.yml的on-run-start钩子，每次运行dbt前自动生成快照：

   on-run-start:
     - "{{ snapshot_source_schemas() }}"
   

2. 对比快照找出变更源
   用Python或shell脚本对比./target/current_source_schema.json和上次的快照文件（比如./target/previous_source_schema.json），找出列有增减的源。

3. 筛选并刷新模型
   同样用dbt ls筛选依赖变更源的增量模型，执行dbt run --full-refresh。

方案三：dbt Cloud用户专属方案 #

如果用dbt Cloud，可以直接开启源Schema变更警报，当源表Schema变化时，触发webhook，在webhook中调用dbt API或命令，对依赖该源的增量模型执行全量刷新，无需自己写脚本。

• 全量刷新必须加--full-refresh参数，确保增量模型重新构建，包含新增的列。
• 每次运行后记得备份当前的状态目录（./target）到./target/previous_state，供下次对比使用。

内容的提问来源于stack exchange，提问作者Sam Clark