我对 Apache Beam/Cloud Dataflow 还是个新手，所以如果我的理解不正确，我深表歉意。
我正在尝试通过管道读取一个大约 30,000 行长的数据文件。我的简单管道首先从 GCS 打开 csv，从数据中提取 header ，通过 ParDo/DoFn 函数运行数据，然后将所有输出写入 csv 回 GCS。该管道有效，是我的第一个测试。
然后我编辑管道以读取 csv、提取标题、从数据中删除标题、通过 ParDo/DoFn 函数运行数据并将标题作为侧输入，然后将所有输出写入一个 csv。唯一的新代码是将 header 作为侧输入传递并从数据中过滤它。
ParDo/DoFn 函数 build_rows 只生成 context.element，这样我就可以确保我的辅助输入正常工作。
我得到的错误如下:
我不确定是什么问题，但我认为这可能是由于内存限制。我将样本数据从 30,000 行减少到 100 行，我的代码终于可以正常工作了。
没有侧输入的管道确实读取/写入所有 30,000 行，但最后我将需要侧输入来对我的数据进行转换。
如何修复我的管道，以便我可以处理来自 GCS 的大型 csv 文件，并且仍然使用侧输入作为文件的伪全局变量？
请您参考如下方法:
我最近编写了一个 CSV file source对于 Apache Beam，我已将其添加到 beam_utils PiPy 包中。具体可以如下使用:
- 安装 beam utils:
pip install beam_utils - 导入:
从 beam_utils.sources 导入 CsvFileSource。 - 将其用作来源:
beam.io.Read(CsvFileSource(input_file))。
在其默认行为中，CsvFileSource 返回由 header 索引的字典 - 但您可以查看文档来决定要使用的选项。
另外，如果你想实现你自己的自定义 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
并且您可以扩展此逻辑以解析 header 和其他特殊行为。
另外，请注意，此源无法拆分，因为它需要按顺序解析，因此它可能代表处理数据时的瓶颈(尽管这可能没问题)。
