如何让dbt Python模型在任意dbt命令执行后（on-run-end）运行？

1. 配置dbt_project.yml的on-run-end钩子 #

在项目根目录的dbt_project.yml中，添加或修改on-run-end配置项，直接指定要执行的Python模型。这个钩子会在任意dbt命令（run/build/test等）执行完成后触发：

on-run-end:
  - "dbt run --select your_python_model_name --target {{ target.name }}"

• 替换your_python_model_name为你的Python模型实际名称
• {{ target.name }}用于保持和当前执行dbt命令时相同的目标环境，避免环境错乱

2. 确保Python模型的依赖逻辑正确 #

你的Python模型需要依赖之前通过宏写入的variant类型表，因此要在模型中明确声明依赖，确保dbt能识别执行顺序：

from snowflake.snowpark import Session
import snowflake.snowpark.functions as F

def model(dbt, session):
    dbt.config(materialized="table")
    # 关联之前存储run_results的variant表
    raw_run_results = dbt.ref("your_run_results_variant_table")
    
    # 这里编写你的数据扁平化及后续处理逻辑
    flattened_df = raw_run_results.select(
        F.col("RUN_ID"),
        F.col("RESULTS:STATUS").alias("EXECUTION_STATUS"),
        F.col("RESULTS:MODEL_NAME").alias("MODEL_NAME"),
        # 根据实际run_results结构添加更多字段
    )
    
    return flattened_df

• 替换your_run_results_variant_table为宏写入的表名
• 可根据需求将materialized设置为incremental，实现增量更新，提升执行效率

3. 验证执行流程 #

• 先手动执行dbt run --select your_python_model_name，确认模型能正常读取variant表并输出正确的扁平化数据
• 执行任意dbt命令（如dbt test），检查on-run-end是否自动触发Python模型运行，最终Snowflake中是否生成处理后的表

可选：用宏触发模型执行（替代shell命令） #

如果环境限制无法直接执行shell命令，可编写宏来触发模型运行：

{% macro trigger_python_model() %}
    {% do run_query("CALL SYSTEM$EXECUTE_IMMEDIATE('dbt run --select your_python_model_name --target " ~ target.name ~ "')") %}
{% endmacro %}

然后在dbt_project.yml中调用该宏：

on-run-end:
  - "{{ trigger_python_model() }}"

注意：这种方法需要Snowflake账户具备执行外部命令的权限，仅适用于特定场景

内容的提问来源于stack exchange，提问作者Mariia Kloss