如何在Apache Beam中统计CSV单列出现次数并修复IndexError错误

• 错误IndexError: list index out of range触发的核心原因是SplitRecords类中手动按逗号分割CSV行后，取索引为1的元素时，分割得到的列表长度不足2。常见的触发场景有三类：
  1. CSV存在脏数据行：比如空行、字段缺失的不完整行，分割后列表长度小于2
  2. 字段内部包含逗号：CSV规范允许被引号包裹的字段内部携带逗号，手动按逗号分割会将单个字段拆分为多个，导致索引整体错位
  3. 换行符转义异常：部分行的换行符存在转义问题，导致ReadFromText读取到的内容不是完整的CSV行

建议使用Python标准库csv做CSV格式解析，同时增加异常捕获逻辑过滤脏数据，避免单条异常行导致整个流水线崩溃，修改后的代码如下：

# 导入CSV解析所需依赖
import csv
from io import StringIO

# Beam相关导入
import apache_beam as beam
from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
import apache_beam.runners.interactive.interactive_beam as ib

class SplitRecords(beam.DoFn):
    """解析CSV行，返回对应city字段"""
    def process(self, element):
        try:
            # 用csv标准库解析，自动处理带引号、内部带逗号的字段
            reader = csv.reader(StringIO(element))
            records = next(reader)
            # 校验字段长度符合要求后再返回city字段（city对应第二列，索引为1）
            if len(records) >= 3:
                return [records[1]]
        except Exception:
            # 异常行直接过滤，也可根据需求添加日志打印
            pass
        return []

p = beam.Pipeline(InteractiveRunner())
lines = p | 'read in file' >> beam.io.ReadFromText("gs://ny-springml-data/AB_NYC_2019.csv", skip_header_lines=1)
records = lines | beam.ParDo(SplitRecords())
groups = (records | beam.Map(lambda x: (x, 1)) | beam.CombinePerKey(sum))
groups | beam.io.WriteToText('TEST2.txt')

# 执行流水线
p.run()

如果后续需要部署到Dataflow生产环境，也可以直接使用Beam原生的CSV读取工具，代码更简洁不易出错：

import apache_beam as beam
from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
from apache_beam.dataframe.io import read_csv

p = beam.Pipeline(InteractiveRunner())
# 直接读取CSV为结构化行，自动识别表头
csv_rows = p | read_csv("gs://ny-springml-data/AB_NYC_2019.csv")
city_count = csv_rows | beam.Map(lambda row: (row.city, 1)) | beam.CombinePerKey(sum)
city_count | beam.io.WriteToText('TEST2.txt')

p.run()

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