目前是否可以使用Apache Beam在python中读取gzip文件？我的管道正在使用这行代码从gcs中提取gzip文件：
beam.io.Read(beam.io.TextFileSource('gs://bucket/file.gz', compression_type='GZIP'))
但是我得到了这个错误：
UnicodeDecodeError: 'utf8' codec can't decode byte 0x8b in position 1: invalid start byte
我们注意到在python光束源代码中，压缩文件似乎是在写入接收器时处理的。https://github.com/apache/incubator-beam/blob/python-sdk/sdks/python/apache_beam/io/fileio.py#L445
更详细的回溯：
Traceback (most recent call last):
File "beam-playground.py", line 11, in <module>
p.run()
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/pipeline.py", line 159, in run
return self.runner.run(self)
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/runners/direct_runner.py", line 103, in run
super(DirectPipelineRunner, self).run(pipeline)
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/runners/runner.py", line 98, in run
pipeline.visit(RunVisitor(self))
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/pipeline.py", line 182, in visit
self._root_transform().visit(visitor, self, visited)
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/pipeline.py", line 419, in visit
part.visit(visitor, pipeline, visited)
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/pipeline.py", line 422, in visit
visitor.visit_transform(self)
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/runners/runner.py", line 93, in visit_transform
self.runner.run_transform(transform_node)
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/runners/runner.py", line 168, in run_transform
return m(transform_node)
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/runners/direct_runner.py", line 99, in func_wrapper
func(self, pvalue, *args, **kwargs)
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/runners/direct_runner.py", line 258, in run_Read
read_values(reader)
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/runners/direct_runner.py", line 245, in read_values
read_result = [GlobalWindows.windowed_value(e) for e in reader]
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/io/fileio.py", line 807, in __iter__
yield self.source.coder.decode(line)
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/coders/coders.py", line 187, in decode
return value.decode('utf-8')
File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/encodings/utf_8.py", line 16, in decode
return codecs.utf_8_decode(input, errors, True)
UnicodeDecodeError: 'utf8' codec can't decode byte 0x8b in position 1: invalid start byte
更新：Python中的TextIOSDK现在支持读取压缩文件。
今天TextIO在Python中SDK实际上不支持从压缩文件中读取。
我遇到了一个类似的问题。我有一个自定义的二进制源，我想解析和获取数据。问题是file.ioAPI基于CSV或ARVO，无论我尝试什么，如果不尝试在换行符时拆分它们，它都不会给我行。你可以想象，二进制文件不能很好地处理这个问题。
起初，我尝试了一个自定义源代码，最终实现了3个类，它复制了核心数据流/光束代码。最后，我编写了这个奇妙的小猴子补丁来完成我需要做的事情（这里是深度源代码测试）。
import apache_beam as beam
from apache_beam.io.fileio import coders
def _TextFileReader__iter(self):
# The full data file is had here and can be read like normal
# You can even limit the character bit here. (I did 9 to grab the file format)
data = self._file.read()
# Now you can either yield the whole file as a single data entry
# and run a ParDo to split it, or you can iterate in here and
# yield each row. I chose the latter, but I'm showing an example
# of the former.
yield data
# This monkeypatch good!
beam.io.fileio.TextFileReader.__iter__ = _TextFileReader__iter
要调用此源并确保它是BINARY，我执行了以下操作：
pipeline | 'start_3' >> beam.io.Read(
beam.io.TextFileSource( 'gs://MY_BUCKET/sample.bin',
coder=coders.BytesCoder()
)
)
注意coders. BytesCoders（）？没有它，它试图将Bytes转换为对我的解析引擎不利的非二进制文件。；）
花了我一整天的时间来解决这个问题。然而，如果你使用这种方法，你几乎可以用数据流中的file.io类做任何事情。；）
我遇到了同样的问题。我试图从GCS读取二进制GZ文件，解压缩它们，然后将它们运送到其他地方进行处理。我分两步解决了这个问题。
首先，确保你使用的是正确的Python库；我的原始库已经过时了（我至少使用了v0.4）：pip install--升级google-blod-dataflow。
其次，我构建了如下管道：
import apache_beam as beam
from apache_beam import (coders, io, transforms)
raw_logs = (p
| io.Read("ReadLogsFromGCS", beam.io.TextFileSource(
"gs://my-bucket/logs-*.gz",
coder=coders.BytesCoder()))
| transforms.Map(lambda x: x)
| io.Write("WriteToLocalhost", io.textio.WriteToText(
"/tmp/flattened-logs",
file_name_suffix=".json")))
p.run()
运行管道后，您应该有一个名为/tmp/flattened-logs. json的文件。
