Apache Beam / GCP Dataflow 编码问题

Question

我在 datalab 中使用 apache 光束/数据流“玩”。我正在尝试从 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 光束之外，我可以通过从 gcs 读取文件来处理此错误：

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

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

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

有什么建议吗？

Answer 1

一个可能的黑客是创建一个从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

并将其作为参数传递给梁提供的 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())

这背后的逻辑详述here

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