使用Apache Beam读取SFTP文件并实现每分钟FTP轮询的方案咨询

你可以根据你的管道类型（流/批）二选一以下方案：

方案1：常驻流管道内置定时触发 #

适合需要持续运行、低延迟处理的场景，直接在现有管道中加定时触发逻辑即可：

• 用Beam内置的PeriodicImpulse（2.30+版本支持）生成每分钟一次的触发信号
• 每个信号触发一次SFTP路径的文件列表拉取，再将文件路径传给你已开发的处理逻辑
• 可选加去重逻辑避免重复处理已经读取过的文件

参考代码示例（Python版）：

import time
import apache_beam as beam
from apache_beam.io.sftp import SftpIO
from apache_beam.transforms.periodicsequence import PeriodicImpulse

with beam.Pipeline(options=你的管道配置) as p:
    # 生成每分钟一次的触发信号
    trigger_signal = p | "每分钟触发" >> PeriodicImpulse(
        fire_interval=60,
        start_timestamp=time.time()
    )
    
    # 拉取SFTP文件列表
    sftp_files = (
        trigger_signal
        | "拉取SFTP文件列表" >> beam.Map(lambda _: SftpIO.list_files(
            host="SFTP主机地址",
            port=22,
            username="登录账号",
            password="登录密码",
            filepattern="/目标路径/*" # 按你的实际文件匹配规则调整
        ))
        | "展开文件列表" >> beam.FlatMap(lambda x: x)
        | "文件去重" >> beam.Distinct() # 避免重复处理未被移除的历史文件
    )
    
    # 此处对接你已经开发完成的后续处理逻辑
    sftp_files | "执行现有处理逻辑" >> 你已实现的处理转换

方案2：批管道配合外部定时调度 #

适合不需要常驻进程、运维更简单的场景，无需修改现有管道代码：

• 直接把你已开发的批处理管道打包为可执行脚本
• 用系统自带的定时调度工具触发每分钟执行一次即可，Linux下可用crontab添加规则：
  * * * * * /usr/bin/python3 /你的脚本存储路径/beam_pipeline.py

• 建议处理完文件后将其移动到SFTP的归档目录、或者直接删除，避免每次拉取都重复拿到已经处理过的文件
• SFTP的连接凭证不要硬编码在代码中，建议通过Beam的PipelineOptions动态传入
• 网络不稳定的场景可以给SFTP操作增加重试配置，避免偶发的连接失败导致管道报错

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