python - 有没有办法使用 ReadFromText 转换(Python)在 Apache Beam 中读取多行 csv 文件？

Question

有没有办法在 Python 中使用 ReadFromText 转换来读取多行 csv 文件？我有一个包含一行的文件，我试图让 Apache Beam 将输入读取为一行，但无法使其正常工作。

def print_each_line(line):
    print line

path = './input/testfile.csv'
# Here are the contents of testfile.csv
# foo,bar,"blah blah
# more blah blah",baz

p = apache_beam.Pipeline()

(p
 | 'ReadFromFile' >> apache_beam.io.ReadFromText(path)
 | 'PrintEachLine' >> apache_beam.FlatMap(lambda line: print_each_line(line))
 )

# Here is the output:
# foo,bar,"blah blah
# more blah blah",baz

尽管多行 csv 文件的标准是将多行元素括在双引号中，但上述代码将输入解析为两行。

Answer 1

Beam 不支持解析 CSV 文件。但是，您可以使用 Python 的 csv.reader。这是一个例子:

import apache_beam
import csv

def print_each_line(line):
  print line

p = apache_beam.Pipeline()

(p 
 | apache_beam.Create(["test.csv"])
 | apache_beam.FlatMap(lambda filename:
     csv.reader(apache_beam.io.filesystems.FileSystems.open(filename)))
 | apache_beam.FlatMap(print_each_line))

p.run()

输出:

['foo', 'bar', 'blah blah\nmore blah blah', 'baz']