如何让DAG中SQL语句识别Airflow的data_interval_start参数？

问题根源 #

你用Python f-string定义SQL语句时，外层的f-string处理让{{ data_interval_start }}变成了普通字符串，Airflow的模板引擎根本没机会识别并替换成实际的时间戳。

解决办法 #

• 直接用普通字符串定义SQL：如果不需要在Python层做变量替换，去掉f-string就行：

  SQL_INSERT_AIRFLOW_TS = "INSERT INTO TRASH.TRASH.TEST_UPDATES SELECT NULL, '{{ data_interval_start }}';"
  

• 确保Operator支持模板渲染：执行SQL的Operator（比如PostgresOperator、BigQueryOperator）必须开启模板支持，而且要把SQL放在它的模板字段里。举个例子：

  from airflow.providers.postgres.operators.postgres import PostgresOperator
  
  insert_ts_task = PostgresOperator(
      task_id="insert_data_interval_start",
      postgres_conn_id="你的数据库连接ID",
      sql="INSERT INTO TRASH.TRASH.TEST_UPDATES SELECT NULL, '{{ data_interval_start }}';"
  )
  

• 混合Python变量与模板变量的正确姿势：如果既要用Python变量又要Airflow模板变量，别用f-string，改用Operator的parameters参数传递：

  TARGET_TABLE = "TRASH.TRASH.TEST_UPDATES"
  insert_ts_task = PostgresOperator(
      task_id="insert_data_interval_start",
      postgres_conn_id="你的数据库连接ID",
      sql="INSERT INTO {{ params.target_table }} SELECT NULL, '{{ data_interval_start }}';",
      parameters={"target_table": TARGET_TABLE}
  )
  

重要提醒 #

只有支持模板的Operator的模板字段里，Airflow的模板变量才会被渲染，用f-string包裹会直接把模板语法当成普通文本输出。

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