beam.io中FileBasedSource的open_file问题在Python 3中出现。

Question

我正在使用CSVRecordSource读取Apache Beam管道中的CSV文件，该管道在read_records函数中使用open_file。Python 2一切正常，但当我迁移到Python 3时，它会抱怨以下问题。

next(csv_reader)
_csv.Error: iterator should return strings, not bytes (did you open the file in text mode?)

默认情况下，open_file方法会使用二进制模式打开文件。

所以我将其更改为使用：

with open(filename, "rt") as f:

但是当我在Google云中运行数据流时，它找不到文件并报错。

FileNotFoundError: [Errno 2] No such file or directory

以下是我的代码

 with self.open_file(filename) as f:
      csv_reader = csv.reader(f, delimiter=self.delimiter, quotechar=self.quote_character)
      header = next(csv_reader)

我该如何在python 3中使用CSVRecordSource？

- tank

---

请告诉我您在哪里使用这个函数？是在 DoFn 中吗？ - Sach

我在 Beam 管道中使用它来读取（CSVRecordSource(input)）。 - tank

Answer 1

你是否在使用这里定义的open_file方法：https://github.com/apache/beam/blob/6f6feaaeebfc82302ba83c52d087b06a12a5b119/sdks/python/apache_beam/io/filebasedsource.py#L166？

如果是的话，我认为你可以直接调用底层的FileSystems.open()，将'application/octet-stream'替换为'text/plain'。

- Yueyang Qiu

Answer 2

我使用iterdecode来迭代解码迭代器提供的输入（字节）来解决这个问题。

csv.reader(codecs.iterdecode(f, "utf-8"), delimiter=self.delimiter, quotechar=self.quote_character)

- tank