Airflow中S3Sensor未找到S3文件时如何跳过DAG剩余任务

方案1：原生S3KeySensor极简配置（无需整合ShortCircuitOperator） #

Airflow 1.10.12及以上版本的S3KeySensor（原S3Sensor）自带soft_fail参数，开启后若传感器超时未检测到指定S3文件，会将自身任务状态标记为SKIPPED而非FAILED。配合下游任务默认的all_success触发规则，所有后续依赖该传感器的任务都会被自动跳过，无需额外编码。
示例代码：

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.operators.dummy import DummyOperator
from airflow import DAG
from datetime import datetime

with DAG(
    dag_id="s3_check_skip_example",
    start_date=datetime(2024,1,1),
    schedule_interval="@daily",
    catchup=False
) as dag:
    s3_check = S3KeySensor(
        task_id="check_s3_file_exists",
        bucket_key="s3://your-bucket/path/to/target_file.csv",
        aws_conn_id="aws_default",
        poke_interval=60, # 每60秒检测一次
        timeout=3600, # 最多检测1小时
        soft_fail=True # 检测失败时标记为SKIPPED而非FAILED
    )

    downstream_task_1 = DummyOperator(task_id="downstream_task_1")
    downstream_task_2 = DummyOperator(task_id="downstream_task_2")

    s3_check >> downstream_task_1 >> downstream_task_2

如果你的DAG里存在不依赖S3Sensor的任务，仅需要跳过S3Sensor之后的分支，给所有需要跳过的下游任务加共同的S3Sensor依赖即可。

方案2：S3检测逻辑与ShortCircuitOperator整合 #

如果你需要更灵活的判断逻辑（比如满足多个条件才跳过、或者需要自定义跳过的返回逻辑），可以直接在ShortCircuitOperator的执行函数里调用S3Hook完成文件检测，完全替代独立的S3Sensor任务，不需要两个算子单独串联。
示例代码：

from airflow.operators.python import ShortCircuitOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.operators.dummy import DummyOperator
from airflow import DAG
from datetime import datetime

def check_s3_file(**context):
    s3_hook = S3Hook(aws_conn_id="aws_default")
    file_exists = s3_hook.check_for_key(
        key="path/to/target_file.csv",
        bucket_name="your-bucket"
    )
    # 返回False时，所有下游任务自动跳过
    return file_exists

with DAG(
    dag_id="s3_shortcircuit_example",
    start_date=datetime(2024,1,1),
    schedule_interval="@daily",
    catchup=False
) as dag:
    s3_check_shortcircuit = ShortCircuitOperator(
        task_id="check_s3_and_skip",
        python_callable=check_s3_file,
        provide_context=True,
        ignore_downstream_trigger_rules=False # 可配置是否忽略下游自定义触发规则
    )

    downstream_task_1 = DummyOperator(task_id="downstream_task_1")
    downstream_task_2 = DummyOperator(task_id="downstream_task_2")

    s3_check_shortcircuit >> downstream_task_1 >> downstream_task_2

注意：如果需要保留部分下游任务即使检测失败也执行，可以将ignore_downstream_trigger_rules设为True，并给不需要跳过的任务设置trigger_rule="all_done"即可。

方案3：自定义整合算子（可选） #

如果需要同时保留S3Sensor的轮询检测能力和ShortCircuit的跳过逻辑，可以自定义一个同时继承两个类的算子，不过实际生产中前两种方案已经可以覆盖99%的使用场景，不需要额外自定义。

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