Airflow DAG复杂调度：利用Timetables实现多间隔与自定义周期

from datetime import datetime
from croniter import croniter

# 尝试使用 */90 作为分钟表达式
it = croniter("*/90 * * * *", datetime(2023, 1, 1))
print(it.get_next(datetime))  # 预期结果可能是 2023-01-01 01:00:00
print(it.get_next(datetime))  # 预期结果可能是 2023-01-01 02:00:00
print(it.get_next(datetime))  # 预期结果可能是 2023-01-01 02:00:00 (注意这里与预期的90分钟间隔不符)

from __future__ import annotations

from datetime import datetime, timedelta

from airflow.timetables.base import DagRunInfo, DataInterval, Timetable
from airflow.utils.state import DagRunState

class CustomComplexTimetable(Timetable):
    """
    一个自定义的Timetable，用于实现复杂的调度逻辑。
    例如，可以结合多个时间间隔，或跳过特定时间。
    """

    def infer_manual_data_interval(self, *, run_after: datetime) -> DataInterval:
        """
        当手动触发DAG时，推断数据间隔。
        """
        # 简单示例：手动触发时，数据间隔为触发时间前一小时
        return DataInterval(start=run_after - timedelta(hours=1), end=run_after)

    def next_dagrun_info(
        self,
        *,
        last_dagrun_info: DagRunInfo | None,
        run_after: datetime,
    ) -> DagRunInfo | None:
        """
        计算并返回下一个DAG Run的调度信息。
        """
        # 示例：实现每90分钟运行，并跳过特定时间（例如，假设不希望在每天的9:00-9:59之间触发）
        # 这个逻辑需要根据具体需求精心设计

        # 如果是首次运行，可以从一个预设的开始时间开始
        if last_dagrun_info is None:
            # 假设从今天的00:00开始
            next_start = run_after.replace(hour=0, minute=0, second=0, microsecond=0)
        else:
            # 从上一个DAG Run的结束时间加上90分钟
            next_start = last_dagrun_info.end + timedelta(minutes=90)

        # 检查是否跳过特定时间
        # 假设我们想跳过所有在9点到9点59分之间开始的运行
        if next_start.hour == 9:
            # 如果下一个计划运行时间落在9点，则跳到10点，并从那里重新计算90分钟
            next_start = next_start.replace(hour=10, minute=0, second=0, microsecond=0)
            # 为了确保90分钟间隔，可能需要更复杂的逻辑，这里仅为示例
            # 实际情况可能需要循环计算直到找到一个有效的时间点

        # 组合多个cron表达式的逻辑也可以在这里实现
        # 例如，可以维护一个预计算的运行时间列表，或者在每次调用时根据多个表达式计算下一个最近的运行时间。

        # 确定数据间隔的结束时间
        next_end = next_start + timedelta(minutes=90) # 假设数据间隔也是90分钟

        # 返回下一个DAG Run的信息
        return DagRunInfo(
            run_after=next_start,
            data_interval=DataInterval(start=next_start, end=next_end),
            # state=DagRunState.SCHEDULED # Airflow会自动设置状态
        )

    def serialize(self):
        """
        将Timetable实例序列化，以便调度器在不同进程间传递。
        """
        return {"__type": "CustomComplexTimetable"} # 简单示例，实际可能需要传递更多参数

from airflow.models.dag import DAG
from datetime import datetime
from custom_timetables import CustomComplexTimetable # 假设你的Timetable类在一个名为 custom_timetables.py 的文件中

with DAG(
    dag_id="my_custom_scheduled_dag",
    start_date=datetime(2023, 1, 1),
    schedule=CustomComplexTimetable(), # 使用你的自定义Timetable实例
    catchup=False,
    tags=["custom_schedule"],
) as dag:
    # ... 你的任务定义 ...
    pass