Airflow S3KeySensor跨多桶监控异常：通配符路径为何不触发？

问题原因 #

你遇到的问题本质是Airflow 1.10.12版本的S3KeySensor在多键列表+通配符+跨桶的组合场景下存在bug：

• 当bucket_key传入完整s3://格式的路径列表时，传感器内部对通配符的解析逻辑会出错，无法正确匹配带*的路径（比如bucket-123的_SUCCESS*）；
• 而不带通配符的路径（bucket-456的_SUCCESS）因为是精确匹配，不受这个bug影响，所以能正常触发。

你的两个猜测部分正确：跨桶轮询本身是支持的，但多键场景下通配符功能确实失效了。

解决方案 #

方案1：拆分独立Sensor任务（推荐） #

直接拆分成两个单独的S3KeySensor，每个对应一个桶的监控需求，再通过trigger_rule配置只要任意一个任务完成就进入下一阶段：

from airflow.operators.dummy_operator import DummyOperator
from airflow.sensors.s3_key_sensor import S3KeySensor

# 监控bucket-123的带通配符文件
wait_for_bucket123 = S3KeySensor(
    task_id='wait_for_bucket123_success',
    bucket_name='bucket-123',
    bucket_key='test-output/_SUCCESS*',
    wildcard_match=True,
    poke_interval=120,
    timeout=600,
    mode='reschedule',
    dag=dag
)

# 监控bucket-456的精确匹配文件
wait_for_bucket456 = S3KeySensor(
    task_id='wait_for_bucket456_success',
    bucket_name='bucket-456',
    bucket_key='test-output/_SUCCESS',
    poke_interval=120,
    timeout=600,
    mode='reschedule',
    dag=dag
)

# 只要任意一个Sensor触发成功，就继续后续任务
proceed_to_next = DummyOperator(
    task_id='proceed_after_success',
    trigger_rule='one_success',
    dag=dag
)

# 任务依赖
wait_for_bucket123 >> proceed_to_next
wait_for_bucket456 >> proceed_to_next

方案2：自定义PythonSensor实现跨桶监控 #

如果不想拆分任务，可以用PythonSensor结合Boto3自己实现跨桶的通配符检查逻辑，完全可控：

from airflow.sensors.python_sensor import PythonSensor
import boto3
from botocore.exceptions import ClientError

def check_success_files():
    s3_client = boto3.client('s3')
    # 定义需要监控的桶和路径规则
    monitor_targets = [
        ('bucket-123', 'test-output/_SUCCESS*'),
        ('bucket-456', 'test-output/_SUCCESS')
    ]
    
    for bucket, key_pattern in monitor_targets:
        if '*' in key_pattern:
            # 处理通配符：先获取前缀，再遍历匹配对象
            prefix = key_pattern.split('*')[0]
            resp = s3_client.list_objects_v2(Bucket=bucket, Prefix=prefix)
            if 'Contents' in resp:
                for obj in resp['Contents']:
                    # 简单匹配_SUCCESS相关文件，可根据需求调整匹配逻辑
                    if '_SUCCESS' in obj['Key']:
                        return True
        else:
            # 精确匹配：检查文件是否存在
            try:
                s3_client.head_object(Bucket=bucket, Key=key_pattern)
                return True
            except ClientError as e:
                # 忽略404错误，其他异常抛出
                if e.response['Error']['Code'] != '404':
                    raise
    # 所有目标都未匹配到
    return False

wait_for_success = PythonSensor(
    task_id='wait_for_success_marker',
    python_callable=check_success_files,
    poke_interval=120,
    timeout=600,
    mode='reschedule',
    dag=dag
)

最佳实践 #

1. 避免旧版本Sensor的复杂组合：Airflow 1.10.x的S3相关Sensor存在不少bug，对于跨桶、通配符这类复杂场景，拆分任务是最稳妥的选择；
2. 利用TriggerRule灵活控制依赖：当需要“满足任意条件即可继续”时，使用trigger_rule='one_success'；如果需要“所有条件都满足”，则用默认的all_success；
3. 优先升级Airflow版本：Airflow 2.x对S3传感器做了大量重构（比如拆分出S3PrefixSensor、S3KeySensor等更细分的组件），修复了旧版本的诸多bug，功能更稳定；
4. 自定义Sensor应对复杂场景：当官方Sensor无法满足需求时，用PythonSensor结合Boto3实现自定义逻辑，灵活性更高。

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