Python Apache Beam 侧输入断言错误

Question

我还是 Apache Beam/Cloud Dataflow 的新手，所以如果我的理解不正确，我深表歉意。

我正在尝试通过管道读取约 30,000 行长的数据文件。我的简单管道首先从 GCS 打开 csv，从数据中提取标头，通过 ParDo/DoFn 函数运行数据，然后将所有输出写入 csv 回 GCS。该管道有效，是我的第一个测试。

然后我编辑了管道以读取 csv，提取标题，从数据中删除标题，通过 ParDo/DoFn 函数运行数据，并将标题作为侧输入，然后将所有输出写入一个.csv。唯一的新代码是将标头作为侧面输入传入并从数据中过滤。

ParDo/DoFn 函数 build_rows 只生成 context.element，以便我可以确保我的辅助输入正常工作。

我得到的错误如下：
我不确定问题是什么，但我认为这可能是由于内存限制。我将示例数据从 30,000 行减少到 100 行，我的代码终于可以工作了。

没有侧输入的管道会读/写所有 30,000 行，但最终我需要侧输入来对我的数据进行转换。

如何修复我的管道，以便我可以处理来自 GCS 的大型 csv 文件，并且仍然使用边输入作为文件的伪全局变量？

Answer 1

我最近为 Apache Beam 编写了 CSV file source，并将其添加到 beam_utils PiPy 包中。具体可以如下使用：

1. 安装梁工具：pip install beam_utils
2. 导入：from beam_utils.sources import CsvFileSource。
3. 将其用作来源：beam.io.Read(CsvFileSource(input_file))。

在其默认行为中，CsvFileSource 返回按标题索引的字典 - 但您可以查看文档来决定要使用的选项。

另外，如果你想实现自己的自定义CsvFileSource，你需要继承Beam的FileBasedSource：

import csv
class CsvFileSource(beam.io.filebasedsource.FileBasedSource):
  def read_records(self, file_name, range_tracker):
    self._file = self.open_file(file_name)
    reader = csv.reader(self._file)
    for i, rec in enumerate(reader):
      yield res

您可以扩展此逻辑以解析标头和其他特殊行为。

另外，作为一个注释，这个源不能被拆分，因为它需要被顺序解析，所以在处理数据时它可能代表一个瓶颈（尽管这可能没问题）。