如何将Dataflow VM生成的本地文件读取至流式Pipeline中处理

你提到的上传GCS再读取的方案可满足持久化需求，但存在额外的远程IO开销，以下是更优的实现方式： #

方案1：直接在DoFn中读取本地文件（无额外IO开销，优先级最高） #

Beam官方提供的IO连接器主要面向分布式、跨节点的统一数据源读取场景，如果你要读取的/tmp/filename.txt是当前工作节点（worker）本地生成的，完全不需要使用专门的IO连接器，直接在DoFn中调用对应编程语言的原生文件读取能力，或者Beam内置的FileSystems工具类读取即可。
示例（Python版本）：

import apache_beam as beam
from apache_beam.io.filesystems import FileSystems

# 方式1：原生文件读取
class ReadLocalFileFn(beam.DoFn):
    def process(self, element):
        # 此处element可为触发读取的信号，比如文件生成完成的标识
        with open('/tmp/filename.txt', 'r', encoding='utf-8') as f:
            for line in f:
                yield line.strip()

# 方式2：使用Beam FileSystems工具类，适配性更强
class ReadLocalFileWithFsFn(beam.DoFn):
    def process(self, element):
        with FileSystems.open('/tmp/filename.txt', 'r') as f:
            for line in f:
                yield line.decode('utf-8').strip()

适用场景：生成文件的步骤和读取处理步骤在同一个worker上执行（中间无GroupByKey、Shuffle类跨节点流转操作）的场景。 #

方案2：跳过落盘直接向下游传递数据 #

如果生成文件内容的逻辑本身就在Pipeline的DoFn中实现，可以直接把生成的内容作为PCollection的元素输出给下游处理，完全不需要写入本地磁盘再读取，省去磁盘IO开销。
适用场景：生成的文件内容不需要持久化留存，仅需当前Pipeline下游处理的场景。 #

方案3：兼容持久化需求的混合方案 #

如果你需要把生成的文件留底供后续其他任务使用，或者担心worker故障导致/tmp路径的临时文件丢失，可以保留上传GCS的逻辑，但可以同步把内容传给下游，不需要等上传完成再读GCS，减少等待开销。 #

• 不要尝试跨worker读取本地文件，Dataflow的worker之间文件系统相互隔离，A节点生成在/tmp下的文件无法被B节点访问
• 流式Pipeline运行过程中worker可能发生扩容、重启或销毁，/tmp路径的临时文件不会持久化，核心数据建议同步写入持久化存储
• 大文件读取要做分批处理，避免单元素内存占用过高引发worker OOM

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