Airflow DAG条件化执行：利用PythonSensor实现特定日期检查

from datetime import datetime, timedelta
import calendar

def is_last_tuesday_of_month(**context):
    """
    检查Airflow的execution_date是否是当前月份的最后一个周二。
    """
    execution_date = context["execution_date"]
    current_year = execution_date.year
    current_month = execution_date.month

    # 获取当前月份的总天数
    # calendar.monthrange(year, month) 返回 (该月第一天的星期几, 该月总天数)
    _, num_days_in_month = calendar.monthrange(current_year, current_month)

    # 构建当前月份的最后一天
    last_day_of_month = datetime(current_year, current_month, num_days_in_month).date()

    # 从月末最后一天开始往前推，直到找到第一个周二
    days_to_subtract = 0
    while (last_day_of_month - timedelta(days=days_to_subtract)).weekday() != calendar.TUESDAY:
        days_to_subtract += 1
        # 安全检查，防止无限循环（虽然通常不会发生，因为每个月至少有一个周二）
        if days_to_subtract > 7:
            print(f"Error: Could not find Tuesday in {current_year}-{current_month}")
            return False

    last_tuesday_date = last_day_of_month - timedelta(days=days_to_subtract)

    print(f"Execution Date: {execution_date.date()}")
    print(f"Last Tuesday of {current_year}-{current_month}: {last_tuesday_date}")

    # 比较执行日期与计算出的月末最后一个周二
    return execution_date.date() == last_tuesday_date

from airflow import DAG
from airflow.sensors.python import PythonSensor
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
import calendar

# 定义上述的 is_last_tuesday_of_month 函数
def is_last_tuesday_of_month(**context):
    """
    检查Airflow的execution_date是否是当前月份的最后一个周二。
    """
    execution_date = context["execution_date"]
    current_year = execution_date.year
    current_month = execution_date.month

    _, num_days_in_month = calendar.monthrange(current_year, current_month)
    last_day_of_month = datetime(current_year, current_month, num_days_in_month).date()

    days_to_subtract = 0
    while (last_day_of_month - timedelta(days=days_to_subtract)).weekday() != calendar.TUESDAY:
        days_to_subtract += 1
        if days_to_subtract > 7:
            print(f"Error: Could not find Tuesday in {current_year}-{current_month}")
            return False

    last_tuesday_date = last_day_of_month - timedelta(days=days_to_subtract)

    print(f"Checking date: {execution_date.date()}")
    print(f"Calculated last Tuesday: {last_tuesday_date}")

    return execution_date.date() == last_tuesday_date

with DAG(
    dag_id='conditional_last_tuesday_dag',
    start_date=datetime(2023, 1, 1),
    schedule_interval='@daily', # 每天运行，但只有在条件满足时才执行下游任务
    catchup=False,
    tags=['example', 'sensor', 'conditional'],
) as dag:
    # 传感器任务：检查是否是月末最后一个周二
    check_last_tuesday = PythonSensor(
        task_id='check_if_last_tuesday',
        python_callable=is_last_tuesday_of_month,
        poke_interval=60 * 5,  # 每5分钟检查一次 (在实际生产中，对于日期检查可以设置更长的间隔或一次性检查)
        timeout=60 * 60 * 24, # 最长等待24小时
        soft_fail=True,       # 如果条件不满足，任务标记为skipped，下游任务也skipped
        mode='poke',          # 默认模式，周期性执行callable
    )

    # 原始任务T1到T5，现在它们将依赖于传感器的成功
    T1 = BashOperator(
        task_id='delete_gcs_files',
        bash_command='echo "Deleting all files from GCS..." && sleep 5',
    )

    T2 = BashOperator(
        task_id='run_sql_query_1',
        bash_command='echo "Running SQL query 1 and outputting to BigQuery..." && sleep 5',
    )

    T3 = BashOperator(
        task_id='run_sql_query_2',
        bash_command='echo "Running SQL query 2 and outputting to BigQuery..." && sleep 5',
    )

    T4 = BashOperator(
        task_id='run_sql_query_3',
        bash_command='echo "Running SQL query 3 and placing CSV in GCS..." && sleep 5',
    )

    T5 = BashOperator(
        task_id='copy_append_history',
        bash_command='echo "Copying and appending reference numbers to history table..." && sleep 5',
    )

    # 定义任务依赖关系
    # 只有当check_last_tuesday传感器成功时，T1及后续任务才会运行
    check_last_tuesday >> T1 >> T2 >> T3 >> T4 >> T5