如何让Apache Beam的ReadFromText安全读取并跳过错误文件

默认的beam.io.ReadFromText遇到损坏的压缩文件时会直接抛出异常终止管道，你可以通过自定义读取逻辑+异常捕获的方式跳过这些错误文件，以下是具体实现方案：

方法一：自定义文件读取函数，逐个处理并跳过错误 #

先获取目标文件列表，再对每个文件单独尝试读取，捕获解压缩相关的异常并跳过：

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

def read_bz2_file_safely(file_path):
    try:
        # 手动处理bz2文件读取，捕获解压缩错误
        with FileSystems.open(file_path) as f:
            with bz2.BZ2File(f, 'rb') as bz2_f:
                for line in bz2_f:
                    yield line.decode('utf-8').strip()
    except (OSError, ValueError) as e:
        # 捕获解压缩错误或无效格式错误，直接跳过该文件
        print(f"Skipping corrupted file {file_path}: {str(e)}")
        return

p1 = beam.Pipeline()

(p1
        # 替换为你的文件路径，支持通配符如'*.bz2'
        | 'List files' >> beam.Create(['bad_file.bz2'])
        | 'Read safely' >> beam.FlatMap(read_bz2_file_safely)
        | 'Write output' >> beam.io.WriteToText('file_out.txt')
 )

p1.run()

方法二：结合ReadAllFromText与TryExcept转换 #

如果你更倾向于使用Beam内置的IO组件，可以用ReadAllFromText配合TryExcept来捕获读取错误：

import apache_beam as beam
from apache_beam.transforms.util import TryExcept

p1 = beam.Pipeline()

(p1
        | 'List files' >> beam.Create(['bad_file.bz2'])
        # 使用ReadAllFromText读取每个文件，指定压缩格式
        | 'Read files' >> beam.io.ReadAllFromText(compression_type='bz2')
        # 捕获读取时的异常，跳过错误文件
        | 'Handle errors' >> TryExcept(
            lambda x: x,
            on_exception=lambda exc, elem: print(f"Skipping file {elem}: {exc}")
        )
        | 'Write output' >> beam.io.WriteToText('file_out.txt')
 )

p1.run()

关键补充 #

• 若需批量处理文件（如用通配符匹配），可通过FileSystems.match自动获取文件列表，替代手动beam.Create：

  from apache_beam.io.filesystems import FileSystems
  
  file_pattern = 'path/to/*.bz2'
  matches = FileSystems.match([file_pattern])
  file_paths = [match_metadata.path for match in matches[0].metadata_list]
  # 后续将file_paths传入beam.Create即可
  

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