character-encoding - Apache Beam/GCP 数据流编码问题

Question

我正在数据实验室中“玩”apache beam/dataflow。 我正在尝试从 gcs 读取一个 csv 文件。 当我使用以下方法创建 pcollection 时:

lines = p | 'ReadMyFile' >> beam.io.ReadFromText('gs://' + BUCKET_NAME + '/' + input_file, coder='StrUtf8Coder')

我收到以下错误:

LookupError: unknown encoding: "THE","NAME","OF","COLUMNS"

似乎列的名称被解释为编码？

我不明白哪里出了问题。 如果我不指定我得到的“编码器”

UnicodeDecodeError: 'utf8' codec can't decode byte 0xe0 in position 1045: invalid continuation byte

在 apache beam 之外，我可以通过从 gcs 读取文件来处理这个错误:

blob = storage.Blob(gs_path, bucket)
data = blob.download_as_string()
data.decode('utf-8', 'ignore')

我读到 apache beam 只支持 utf8 并且文件不只包含 utf8。

我应该下载然后转换为 pcollection 吗？

有什么建议吗？

Answer 1

一个可能的 hack 是创建一个继承自 Coder 类 ( apache_beam.coders.coders.Coder ) 的类

from apache_beam.coders.coders import Coder

class ISOCoder(Coder):
    """A coder used for reading and writing strings as ISO-8859-1."""

    def encode(self, value):
        return value.encode('iso-8859-1')

    def decode(self, value):
        return value.decode('iso-8859-1')

    def is_deterministic(self):
        return True

并将其作为参数传递给 beam 提供的 ReadFromText IO 转换 (apache_beam.io.textio.ReadFromText) 像这样

from apache_beam.io import ReadFromText

with beam.Pipeline(options=pipeline_options) as p:  
    new_pcollection = (  p | 'Read From GCS' >>
               beam.io.ReadFromText('input_file', coder=ISOCoder())

这背后的逻辑在这里详述

https://medium.com/@khushboo_16578/cloud-dataflow-and-iso-8859-1-2bb8763cc7c8