Airflow中BigQueryCheckOperator如何传入DAG execution_date参数

你当前的写法有两个核心问题：

1. 直接在Operator初始化时用.format()拼接datetime.now()生成SQL，这个日期会在Airflow解析加载DAG文件的时候就被计算为固定值，既不是DAG调度的执行日期，还会因为DAG解析时间和任务实际运行时间不一致出现日期错误。
2. 你写的retry_delay=10参数类型不符合要求，Airflow的重试间隔需要传入timedelta类型，直接传整数会触发参数报错。

BigQueryCheckOperator的sql参数本身原生支持Jinja模板渲染，不需要提前用Python语法做字符串拼接，直接在SQL里引用Airflow内置的模板变量即可，模板变量会在任务实际运行时被替换为真实的执行上下文值。
Airflow内置了现成的ds变量，就是已经格式化为YYYY-MM-DD格式的执行日期字符串（UTC时区），可以直接使用，修改后的完整代码如下：

from datetime import timedelta
from airflow.providers.google.cloud.operators.bigquery import BigQueryCheckOperator

BigQueryCheckOperator(
    task_id='mytask',
    retry_delay=timedelta(seconds=10),
    retries=2,
    sql="""
       SELECT
        COUNT(*)
       FROM
        mydataset.mytable AS tb
       WHERE
        DATE(tb.date) = '{{ ds }}'
    """,
    bigquery_conn_id='my_google_cloud_conn',
    dag=dag
)

如果需要对执行日期做自定义处理（比如转换时区、调整日期偏移），可以直接在模板里调用execution_date（是Pendulum时间对象，支持原生时间操作），常用的写法包括：

• 转东八区日期：{{ execution_date.in_timezone('Asia/Shanghai').strftime('%Y-%m-%d') }}
• 取前一天日期：{{ prev_ds }} 或者 {{ (execution_date - macros.timedelta(days=1)).strftime('%Y-%m-%d') }}
• 不带横杠的日期格式：{{ ds_nodash }}（对应格式YYYYMMDD）

注意：Airflow 2.2+ 版本中也可以用logical_date替代execution_date，两者在常规调度场景下取值完全一致，内置的ds等日期变量会自动适配，不需要额外修改。

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