Airflow中使用BranchDateTimeOperator遇时区时间对比错误

错误原因 #

你遇到的TypeError: can't compare offset-naive and offset-aware datetimes，本质是无时区感知的datetime对象和带时区感知的datetime对象无法直接比较。你用pendulum.datetime()创建的target_upper、target_lower属于无时区感知类型，而Airflow内部用来做比较的now是带时区的（默认继承DAG时区或系统时区）。

解决方案 #

给你创建的pendulum时间对象指定明确时区，确保和Airflow使用的时区一致即可，有两种常用方式：

方式一：直接在pendulum.datetime()中指定时区

在初始化时间时通过tz参数传入时区标识（比如UTC、Asia/Shanghai等）：

target_upper=pendulum.datetime(2023, 6, 21, 20, 0, 0, tz="UTC"),
target_lower=pendulum.datetime(2023, 6, 21, 13, 0, 0, tz="UTC"),

方式二：先定义时区对象，再创建时间

先初始化对应时区的pendulum对象，再用它来生成时间：

# 定义时区，和DAG时区保持一致
dag_timezone = pendulum.timezone("Asia/Shanghai")

target_upper = dag_timezone.datetime(2023, 6, 21, 20, 0, 0)
target_lower = dag_timezone.datetime(2023, 6, 21, 13, 0, 0)

修正后的完整代码示例 #

import pendulum
from airflow import DAG
from airflow.operators.datetime import BranchDateTimeOperator
from airflow.operators.dummy import DummyOperator

# DAG默认参数，指定时区确保一致性
default_args = {
    "start_date": pendulum.datetime(2023, 6, 20, tz="UTC"),
    "timezone": "UTC"
}

with DAG(
    dag_id="datetime_branch_dag",
    default_args=default_args,
    schedule_interval="@daily",
    catchup=False
) as new_dag:
    # 定义任务（示例用DummyOperator，实际替换为你的任务）
    sample_pybq1 = DummyOperator(task_id="sample_pybq1")
    sample_pybq12 = DummyOperator(task_id="sample_pybq12")
    sample_bq_task = DummyOperator(task_id="sample_bq_task")

    cond3 = BranchDateTimeOperator(
        task_id="datetime_branch",
        follow_task_ids_if_true=["sample_pybq1"],
        follow_task_ids_if_false=["sample_pybq12"],
        # 给时间指定时区，和DAG时区匹配
        target_upper=pendulum.datetime(2023, 6, 21, 20, 0, 0, tz="UTC"),
        target_lower=pendulum.datetime(2023, 6, 21, 13, 0, 0, tz="UTC"),
        dag=new_dag,
    )

    cond3 >> [sample_pybq1, sample_pybq12] >> sample_bq_task

注意事项 #

务必保证target_upper、target_lower的时区和DAG的timezone参数（或Airflow全局默认时区）一致，否则仍会出现时区类型不匹配的问题。

内容的提问来源于stack exchange，提问作者Mani Shankar.S