能否指定DBT Snapshot运行时时间？Airflow重跑快照时间字段异常

dbt快照默认会用数据库的当前时间生成DBT_VALID_FROM和DBT_VALID_TO字段，重跑时自然取执行时刻的时间。要指定快照更新的运行时间，完全可以通过以下几种方式实现：

1. 自定义变量覆盖时间逻辑 #

在dbt项目中定义全局变量，然后在快照模型里直接引用这个变量生成时间字段：

• 在dbt_project.yml中添加变量：

  vars:
    snapshot_run_time: '2024-05-20 14:30:00' # 替换为你需要的指定时间
  

• 修改快照文件中的时间字段赋值逻辑（以timestamp策略为例）：

  {{ config(
    strategy='timestamp',
    unique_key='your_unique_key',
    updated_at='updated_at_column'
  ) }}
  
  select
    *,
    {{ var('snapshot_run_time') }} as dbt_valid_from,
    case when dbt_valid_to is not null then {{ var('snapshot_run_time') }} else null end as dbt_valid_to
  from source('your_source_schema', 'your_source_table')
  

  如果用check策略，同样需要把时间字段替换为变量值。

2. 通过Airflow传递运行时间 #

在Airflow的dbt任务中，通过--vars参数直接传递指定时间（比如用DAG的execution_date）：

from airflow.operators.bash import BashOperator

dbt_snapshot_task = BashOperator(
    task_id='run_dbt_snapshot',
    bash_command='dbt snapshot --vars "{snapshot_run_time: \'{{ execution_date.strftime(\'%Y-%m-%d %H:%M:%S\') }}\'}"',
    dag=dag
)

这里用execution_date作为快照的运行时间，也可以替换成固定时间字符串或其他Airflow变量。

3. 覆盖dbt内置宏（进阶方案） #

通过自定义宏替换dbt默认的时间获取逻辑，让所有快照统一使用指定时间：

• 在项目的macros/目录下创建snapshot_time_override.sql：

  {% macro snapshot_get_time() %}
    {{ return(var('snapshot_run_time', current_timestamp())) }}
  {% endmacro %}
  

  这个宏会优先使用snapshot_run_time变量的值，没有指定时 fallback 到当前时间。之后所有快照的DBT_VALID_FROM和DBT_VALID_TO都会自动使用变量指定的时间。

注意事项 #

• 指定的时间格式必须和目标数据库兼容，比如PostgreSQL用'YYYY-MM-DD HH24:MI:SS'，MySQL用'YYYY-MM-DD HH:MI:SS'。
• 重跑时要保证指定时间晚于上一次快照的时间，避免出现时间线逻辑混乱。
• 增量快照场景下，要确认更新条件和指定时间的配合，防止数据漏更或错误覆盖。

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