apache-beam - 如何使用Beam读取大型CSV？

Question

我试图弄清楚如何使用Apache Beam读取大型CSV文件。 “大”是指几千兆字节(因此一次将整个CSV读取到内存中是不切实际的)。

到目前为止，我已经尝试了以下选项:

使用TextIO.read():这不好，因为带引号的CSV字段可能包含换行符。此外，这会尝试将整个文件立即读取到内存中。

编写一个DoFn，将其作为流读取文件并发出记录(例如，使用commons-csv)。但是，这仍然一次读取整个文件。

尝试使用SplittableDoFn as described here。我的目标是使它逐渐作为Unbounded PCollection发出记录-基本上是将我的文件转换为记录流。但是，(1)很难正确计算计数；(2)由于ParDo创建了多个线程，因此需要进行一些骇人听闻的同步；(3)我得到的PCollection仍然不受限制。

尝试创建我自己的UnboundedSource。这似乎非常复杂，而且文献记载很少(除非我遗漏了什么？)。

Beam是否提供任何简单的方法来允许我以自己想要的方式解析文件，而不必在进行下一个转换之前将整个文件读入内存？

Answer 1

从Beam的预期角度来看，TextIO应该做正确的事情，即尽可能快地读取文本文件并将事件发送到下一个阶段。

我猜您正在为此使用DirectRunner，这就是为什么您看到大量内存占用的原因。希望这不是太多的解释:DirectRunner是用于小型作业的测试运行器，因此可以将中间步骤缓冲在内存中，而不是磁盘中。如果仍在测试管道，则应使用一小部分数据样本，直到您认为它可以正常工作为止。然后，您可以使用Apache Flink运行程序或Google Cloud Dataflow运行程序，它们都将在需要时将中间阶段写入磁盘。