如何用Python实现Java Dataflow流式代码中GCS桶新文件轮询功能

方案1：内置组件实现轮询（无额外服务依赖） #

直接用Beam Python原生API组合实现10秒间隔轮询GCS新文件的逻辑，核心是用定时触发+文件去重+批量读文件，完全对齐Java版本的功能逻辑：

import apache_beam as beam
from apache_beam.io import ReadAllFromText
from apache_beam.io.gcp.gcsio import GcsIO
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.transforms import PeriodicImpulse
from apache_beam.transforms.userstate import SetStateSpec, StateSpec
from apache_beam.utils.timestamp import Duration
import typing

# 去重DoFn：存储已处理的文件路径，避免重复消费
class DeduplicateFilesFn(beam.DoFn):
    PROCESSED_FILES = StateSpec('processed_files', SetStateSpec(typing.Text))
    # 设1天超时清理状态，避免状态无限膨胀，可根据业务调整
    EXPIRE_TIME = Duration.of(days=1)

    def process(self, element, processed_files=beam.DoFn.StateParam(PROCESSED_FILES)):
        file_path = element
        # 检查是否已经处理过
        if file_path not in processed_files.read():
            processed_files.add(file_path)
            yield file_path

# 管道启动配置
options = PipelineOptions()
options.view_as(StandardOptions).streaming = True
gcs_target_path = "gs://abc/xyz/*"  # 匹配目标路径下所有文件

with beam.Pipeline(options=options) as p:
    # 1. 每10秒触发一次轮询，永不停止，对应Java代码的Watch.Growth.never()
    file_content = (p
             | "Trigger every 10s" >> PeriodicImpulse(
                 fire_interval=10, stop_timestamp=float('inf'))
             | "List GCS files" >> beam.FlatMap(
                 lambda _: GcsIO().list_prefix(gcs_target_path).keys())
             | "Deduplicate processed files" >> beam.ParDo(DeduplicateFilesFn())
             | "Read file content" >> ReadAllFromText()
             )
    # 后续接你的数据转换、加载逻辑
    # file_content | "Transform data" >> ... | "Load to destination" >> ...

方案2：GCS事件通知联动Pub/Sub（生产级推荐） #

轮询方式在文件量大时性能较低，推荐用GCS原生事件通知能力，实现新文件产生后立即触发处理，无需主动轮询：

1. 提前给目标GCS桶配置事件通知，触发事件选择OBJECT_FINALIZE（文件上传完成事件），通知目标指定为你提前创建的Pub/Sub主题
2. 管道直接订阅该Pub/Sub主题获取新上传的文件路径，再读取文件内容即可：

import apache_beam as beam
from apache_beam.io import ReadFromPubSub, ReadAllFromText
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
import json

options = PipelineOptions()
options.view_as(StandardOptions).streaming = True
pubsub_subscription = "projects/你的项目ID/subscriptions/你的订阅名"

with beam.Pipeline(options=options) as p:
    file_content = (p
             | "Read GCS event from PubSub" >> ReadFromPubSub(subscription=pubsub_subscription)
             | "Parse file path" >> beam.Map(lambda msg: json.loads(msg)['name'])
             | "Splice full GCS path" >> beam.Map(lambda file_name: f"gs://abc/xyz/{file_name}")
             | "Read file content" >> ReadAllFromText()
             )
    # 后续接你的数据转换、加载逻辑

注意事项 #

• 方案1的状态存储依赖Beam运行器的状态后端，生产运行时建议开启持久化状态存储，避免重启后重复处理历史文件
• 若业务允许少量重复处理，可去掉去重逻辑，简化实现
• 若需要控制管道停止时机，可修改PeriodicImpulse的stop_timestamp参数为指定的时间戳

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