咨询Airflow Sensor对未完全加载文件的处理逻辑

Airflow的基础文件类传感器（比如FileSensor）不会主动判断文件是否完全复制完成，它的默认逻辑仅检查目标路径下是否存在对应文件名的实体。如果大文件还在复制过程中（操作系统已创建文件占位符但内容未写入完毕），FileSensor会直接判定文件存在，触发后续任务，此时读取的就是不完整的文件内容。

原因说明 #

操作系统在执行大文件复制时，通常会先创建一个空文件（或占位文件），再逐步写入内容。FileSensor的核心校验逻辑仅针对「文件是否存在」，不会校验文件大小是否稳定、写入是否完成，因此无法识别未完成复制的文件。

解决方法 #

要避免读取未完全加载的文件，可通过以下方式实现：

• 自定义传感器校验文件大小稳定性：在检测到文件存在后，间隔一段时间再次检查文件大小，若两次大小一致则判定文件写入完成。示例代码：

from airflow.sensors.base import BaseSensorOperator
import os
from airflow.utils.decorators import apply_defaults

class StableFileSensor(BaseSensorOperator):
    @apply_defaults
    def __init__(self, filepath, poke_interval=5, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.filepath = filepath
        self.poke_interval = poke_interval
        self.prev_size = -1

    def poke(self, context):
        if not os.path.exists(self.filepath):
            return False
        current_size = os.path.getsize(self.filepath)
        if current_size == self.prev_size:
            return True
        self.prev_size = current_size
        return False

• 采用原子性文件写入方式：复制文件时先写入临时文件名（如data.tmp），待复制完成后再重命名为目标文件名（如data.csv）。操作系统的重命名操作是原子性的，FileSensor监听目标文件名时，只会在文件完全复制完毕后才会检测到其存在。
• 使用增强型传感器：部分特定存储的传感器支持完整性校验，比如针对S3的S3KeySensor可配置相关参数验证文件完整性，部分第三方Airflow插件也提供了带完整性校验的传感器实现。

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