如何使用Airflow的FileSensor glob匹配指定字符串+后缀的文件？

核心需求匹配方案 #

要匹配**包含指定字符串（如example_file）且后缀为.txt/.csv**的文件，先分析你给出的两种写法，再给出正确实现：

你的两种写法可行性分析 #

1. example_file**.txt
   这种写法不可行。在glob规则中，**用于匹配任意层级的子目录（需配合recursive=True生效），而非匹配多字符。使用该模式会意外匹配到子目录中名称以example_file开头的.txt文件，不符合需求。

2. example_file*.txt
   这种写法完全可行。glob中的*用于匹配任意长度的非路径分隔符字符（包括0个），可以精准匹配像example_file_2023-01-01.txt这类名称包含example_file且后缀为.txt的文件。

同时匹配.txt和.csv的实现 #

Airflow的FileSensor底层依赖Python的glob模块，Python原生glob不支持直接写OR条件（如*.{txt,csv}这种bash风格扩展），推荐两种实现方式：

方式1：传入文件路径列表

直接给filepath参数传入包含两种后缀匹配规则的列表，FileSensor会检测任意一个规则匹配到文件即触发：

file_check = FileSensor(
    task_id='file_check',
    filepath=[
        'filepath/to/folder/example_file*.txt',
        'filepath/to/folder/example_file*.csv'
    ]
)

方式2：自定义匹配逻辑（进阶）

如果需要更灵活的匹配规则，可以继承FileSensor并重写poke方法，使用fnmatch实现复杂匹配：

from airflow.providers.filesystem.sensors.file import FileSensor
import fnmatch
import os

class CustomFileSensor(FileSensor):
    def poke(self, context):
        base_dir = os.path.dirname(self.filepath)
        for filename in os.listdir(base_dir):
            if fnmatch.fnmatch(filename, 'example_file*') and filename.endswith(('.txt', '.csv')):
                return True
        return False

# 使用自定义传感器
file_check = CustomFileSensor(
    task_id='file_check',
    filepath='filepath/to/folder/'
)

关键规则总结 #

• Airflow FileSensor的glob匹配遵循Python glob模块规则：
  • *：匹配任意非路径分隔符的字符（0或多个）
  • **：仅在recursive=True时匹配任意层级目录，不用于匹配文件名
  • 不支持bash风格的{}扩展匹配多后缀，需用列表或自定义逻辑实现

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