在airflow 2.2中,时间表安排(Timetable)是在代码中动态注册的。这意味着,在调度作业时,必须以某种方式首先将时间表注册到Airflow中,否则会导致错误“时间表未注册”。
以下是一个例子,可以解决这个问题:
首先,在代码中添加如下行:
from airflow.plugins_manager import AirflowPlugin from my_plugin.timetables import MyTimetable
接着,定义MyPlugin类并继承AirflowPlugin:
class MyPlugin(AirflowPlugin): name = "my_plugin" timetables = [MyTimetable()]
最后,将该插件安装到Airflow中:
pip install /path/to/my_plugin/
执行以上三个步骤后,即可将时间表成功注册到Airflow中,并解决错误“时间表未注册”的问题。