这个问题通常出现在Airflow 1.9之后的版本中,其中deferred模块被替换为defer模块。为了解决这个问题,需要将任务代码中的deferred传感器更改为defer传感器。
例如,将以下代码:
from airflow.operators.sensors import TimeDeltaSensor from airflow.utils import dates
... sensor = TimeDeltaSensor( task_id='time_delta_sensor', delta=datetime.timedelta(minutes=-10), dag=dag)
更改为:
from airflow.sensors.time_delta import TimeDeltaSensor from airflow.utils import dates
... sensor = TimeDeltaSensor( task_id='time_delta_sensor', delta=datetime.timedelta(minutes=-10), dag=dag)
这将使用新的defer传感器,并应该解决AttributeError __aenter__错误。