Apache Beam / GCP 数据流编码问题

Question

我 "playing" 在数据实验室中使用 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"

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

我不明白哪里出了问题。如果我不指定 "coder" 我得到

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

我建议更改实际文件的编码。如果使用 "Save as" 保存文件，则可以 select UTF-8 编码格式 excel CSV 和常规 .txt。完成后，您需要确保添加一行代码，例如

class DoWork(beam.DoFn):
  def process(self, text):
    text = textfilePcollection.encode('utf-8')

    Do other stuff

这不是我想要的方式，因为它不是以代码为中心的，但它以前对我有用。不幸的是，我没有以代码为中心的解决方案。

Answer 2

一个可能的破解方法是创建一个继承自 Coder class (apache_beam.coders.coders.Coder)

的 class

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

Apache Beam / GCP 数据流编码问题

Apache Beam / GCP Dataflow encoding issue

character-encoding

google-cloud-dataflow

apache-beam