Airflow 2.9.0自定义SFTP算子出现‘Detected zombie job’报错求助

针对处理大量SFTP文件时3分钟左右触发僵尸任务报错的问题，以下是几个可行的解决方向：

1. 调整僵尸任务检测阈值 #

Airflow默认的scheduler_zombie_task_threshold参数为180秒（3分钟），若调度器在此时间内未收到任务心跳，就会判定为僵尸任务。你可以在airflow.cfg中增大这个值：

[scheduler]
scheduler_zombie_task_threshold = 300  # 改为5分钟，可根据任务实际耗时灵活调整

同时确认task_heartbeat_sec参数（默认5秒）保持合理值，确保任务有足够频率发送心跳。

2. 在自定义算子中主动触发心跳 #

频繁打印日志或添加短休眠并不等同于发送Airflow任务心跳。你需要在自定义算子的文件处理循环中，主动调用任务实例的心跳方法，明确告知调度器任务仍在运行。示例代码如下：

from airflow.models.baseoperator import BaseOperator
from airflow.providers.sftp.hooks.sftp import SFTPHook

class CustomSFTPTransferOperator(BaseOperator):
    def execute(self, context):
        sftp_hook = SFTPHook(ssh_conn_id='your_sftp_conn')
        files_to_transfer = self.get_files_list()  # 你的文件列表获取逻辑
        
        for idx, file in enumerate(files_to_transfer):
            # 执行SFTP文件传输逻辑
            sftp_hook.get(file, f'/local/path/{file}')
            
            # 每处理10个文件触发一次心跳，可根据文件大小调整间隔
            if idx % 10 == 0:
                self.task_instance.heartbeat()

3. 优化SFTP传输逻辑 #

避免单文件循环处理的高频操作，改用批量传输方式减少循环次数，同时在批量操作间隙触发心跳。比如利用SFTP的批量下载/上传功能，减少单次操作的阻塞时间，让任务有更多机会发送心跳。

4. 检查Worker资源状态 #

大量文件传输可能导致Worker进程CPU、内存耗尽，无法正常发送心跳。监控Worker节点的资源使用率，确认是否存在资源瓶颈，必要时扩容Worker或调整任务的资源分配（如task_memory、task_cpu参数）。

5. 改用官方SFTP组件 #

Airflow官方的SFTPOperator和SFTPHook已内置心跳处理逻辑，稳定性更高。如果你的自定义算子逻辑可通过官方组件实现，建议直接替换，避免重复造轮子带来的问题。

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