如何在Apache Beam/Dataflow的GCS CSV处理中实现空文件检查？

你的问题根源在于：当CSV文件为空时，Beam的ReadFromText不会输出任何元素，所以你的parse_csv函数根本不会被调用，自然触发不了异常。下面提供两种可行的解决方案：

方法一：管道启动前检查文件大小 #

直接在构建管道前，通过Beam的FileSystem API遍历目标文件，检查文件是否为空。这种方法能提前终止管道，避免不必要的资源消耗。

示例代码：

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

# 定义你的输入文件路径/匹配规则
input_pattern = "gs://your-input-bucket/path/*.csv"

# 获取所有匹配的文件元数据
matched_files = FileSystems.match([input_pattern])[0].metadata_list

# 遍历检查每个文件
for file_meta in matched_files:
    # 若文件大小为0，直接抛出异常终止管道
    if file_meta.size_in_bytes == 0:
        raise ValueError(f"检测到空CSV文件：{file_meta.path}")

# 后续正常构建并运行管道
with beam.Pipeline(options=pipeline_options) as p:
    rows = p | "读取CSV" >> beam.io.ReadFromText(input_pattern)
    # ...你的转换和写入逻辑

如果你的CSV文件包含表头（空数据指表头外无内容），可以调整判断逻辑：比如先读取表头的字节数，再对比文件大小是否仅等于表头大小。

方法二：管道运行中统计行数并检查 #

在管道流程中加入全局计数步骤，若读取到的行数为0（或减去表头后为0），则抛出异常让管道失败。这种方法适合需要先读取文件内容再判断的场景。

示例代码：

import apache_beam as beam

def validate_row_count(count):
    # 若行数为0，抛出异常
    if count == 0:
        raise Exception("所有CSV文件均无数据")
    # 如果有表头，可改为 if count <= 1:（假设表头占1行）
    return count

with beam.Pipeline(options=pipeline_options) as p:
    # 读取CSV文件
    rows = p | "读取CSV" >> beam.io.ReadFromText(input_pattern)
    
    # 全局统计行数
    row_count = rows | "统计行数" >> beam.combiners.Count.Globally()
    
    # 校验行数，为空则触发异常
    row_count | "校验数据存在性" >> beam.Map(validate_row_count)
    
    # 原有的解析和转换逻辑
    parsed_rows = rows | "解析CSV" >> beam.Map(parse_csv)
    # ...后续写入GCS的逻辑

为什么你的原有代码无效？ #

你的parse_csv函数处理的是每一行输入元素，但空CSV文件不会生成任何行元素，这个函数根本不会被执行，所以里面的异常永远不会触发。

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