如何配置Airflow GCS传感器任务，3天内检测月度目标文件

核心参数配置思路 #

• 调度触发：设置schedule_interval为"0 0 6 * *"，确保每月6号零点启动DAG
• 检测频率：poke_interval设为3600秒（1小时），平衡检测及时性与API调用成本
• 超时控制：timeout设为3*24*60*60秒（3天），刚好覆盖6号到9号的检测窗口，超时则任务失败

完整DAG代码实现 #

from airflow import DAG
from airflow.providers.google.cloud.sensors.gcs import GoogleCloudStorageObjectSensor
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 0,
    'start_date': datetime(2024, 1, 1),
}

with DAG(
    'monthly_gcs_file_check',
    default_args=default_args,
    schedule_interval='0 0 6 * *',  # 每月6号零点触发
    catchup=False,
    tags=['gcs', 'monthly']
) as dag:

    check_gcs_file = GoogleCloudStorageObjectSensor(
        task_id='check_gcs_target_file',
        bucket='your-target-bucket',  # 替换为实际存储桶名
        object='path/to/your/target-file.csv',  # 替换为实际文件路径
        poke_interval=3600,  # 每小时检测一次
        timeout=3*24*60*60,  # 3天后超时失败
        mode='poke',  # 使用poke模式，持续检测直到超时或找到文件
        soft_fail=False,  # 超时直接标记任务失败
    )

    check_gcs_file

方案合理性说明 #

• 每小时检测的频率既不会因为过于频繁导致GCS API调用过载，也能保证在文件到达后的1小时内及时捕获
• 3天的超时窗口精准覆盖6号到9号的时间范围，避免提前终止或无意义的长时间等待
• 采用poke模式而非reschedule模式，避免频繁创建销毁任务实例，更适合固定窗口的检测场景

内容的提问来源于stack exchange，提问作者unnest_me