Airflow S3传感器持续运行问题及特定DAG逻辑实现咨询

看起来你遇到了Airflow S3传感器无限轮询的问题，而且想要实现「S3存在CSV文件就继续执行后续EC2任务，否则直接终止DAG」的逻辑对吧？咱们一步步拆解问题，找到最合适的解决方案～

为什么你的S3KeySensor会一直跑？ #

你当前使用的S3KeySensor默认是mode='poke'模式，它会按照默认60秒的poke_interval持续轮询S3，直到找到匹配的文件，或者达到execution_timeout设置的超时时间。但你没有配置execution_timeout，所以它会无限循环下去。

另外你设置的retries=1其实不起作用——这个参数是针对任务失败后的重试次数，而传感器的轮询失败不算任务失败，只会继续下一次轮询，所以不会触发重试逻辑。

解决方案：两种实现思路 #

根据你的需求（每小时调度，仅检查一次文件是否存在，存在就继续），推荐两种方案：

方案一：调整S3KeySensor参数，限制轮询时长

如果还是想用传感器，可以给它加上超时时间，让它在指定时间内没找到文件就停止，结合soft_fail=True标记任务为skipped，后续任务就不会执行：

from datetime import timedelta

check_for_new_csv = S3KeySensor(
    task_id='check_for_new_csv',
    bucket_name='bucket-data',
    bucket_key='*.csv',
    wildcard_match=True,
    soft_fail=True,
    retries=0,  # 无需重试，我们通过超时控制停止
    execution_timeout=timedelta(minutes=5),  # 5分钟内没找到就停止轮询
    poke_interval=60,  # 每分钟检查一次
    mode='poke'
)

这样，超时后传感器任务会被标记为skipped，后续的start_instance因为默认trigger_rule=all_success，会自动跳过，DAG就会正常结束。

方案二：用ShortCircuitOperator替代传感器（更贴合你的需求）

传感器的设计初衷是等待条件满足，而你的需求是一次性检查是否存在，用ShortCircuitOperator结合S3Hook会更高效，只执行一次检查：

from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.operators.python import ShortCircuitOperator

def check_csv_exists():
    # 初始化S3Hook，aws_conn_id替换成你Airflow中配置的AWS连接ID
    s3_hook = S3Hook(aws_conn_id='aws_default')
    # 检查bucket中是否存在匹配的CSV文件
    return s3_hook.check_for_wildcard_key('*.csv', bucket_name='bucket-data')

check_for_new_csv = ShortCircuitOperator(
    task_id='check_for_new_csv',
    python_callable=check_csv_exists,
    soft_fail=True  # 没找到文件时标记任务为skipped，DAG状态为success；设为False则任务失败
)

这个任务的逻辑很清晰：

• 找到CSV文件：返回True，后续start_instance任务正常执行
• 没找到CSV文件：返回False，任务标记为skipped，所有下游任务自动跳过，DAG直接结束

小提示 #

• 用方案二时，要确保Airflow的Connections里配置了正确的AWS连接（有访问目标S3桶的权限）
• 如果希望没找到文件时DAG状态显示为failed，把ShortCircuitOperator的soft_fail设为False即可

备注：内容来源于stack exchange，提问作者elkarel