Python中的Apache Beam-我们可以编写文件页脚吗答案

【问题标题】：Apache Beam in Python- can we write a file footerPython中的Apache Beam-我们可以编写文件页脚吗
【发布时间】：2020-09-21 05:46:01
【问题描述】：

当我们在 Apache Beam 中写入文件时，会使用以下类的 header 参数写入文件头

class apache_beam.io.textio.WriteToText(file_path_prefix, file_name_suffix='', append_trailing_newlines=True, num_shards=0, shard_name_template=None, coder=ToStringCoder, compression_type='auto', header=None)

虽然没有开箱即用的页脚，但有人可以提供一种可以将页脚添加到编写的文件的技术吗？

预期的输出是

Header
Row 1
Row 2
..
..
Row 1000000
Footer

【问题讨论】：

你看过apache_beam.io.filebasedsink模块吗？文档指出您可能需要调整 open() 和/或 close() 函数来自定义文件处理或写入页眉/页脚。
@Peter 该解决方案有效，并且在下面的答案中指出了实现。我们需要覆盖 Sink 并指向自定义编写器中的接收器。

标签： python-3.x google-cloud-dataflow apache-beam

【解决方案1】：

我找到了解决方案。我们需要通过扩展 2 个类来做到这一点

扩展 _TextSink 并覆盖 close() 方法。

    def close(self, file_handle):
    if self._footer is not None:
        self.write_record(file_handle, self._footer)
    file_handle.close()

扩展PTransform 类来编写你自己的作家

class MyWriteToText(PTransform):
def __init__(
        self, file_path_prefix, file_name_suffix='', append_trailing_newlines=True,
        num_shards=0, shard_name_template=None, coder=coders.ToStringCoder(),
        compression_type=CompressionTypes.AUTO, header=None, footer=None):
    self._sink = _TheExtendedSink(
        file_path_prefix, file_name_suffix, append_trailing_newlines,
        num_shards, shard_name_template, coder=coders.ToStringCoder(),
        compression_type=compression_type, header=header, footer=footer)

def expand(self, pcoll):
    return pcoll | Write(self._sink)

【讨论】：

【解决方案2】：

目前 Beam Java SDK 似乎支持此功能，但 Beam Python SDK 不支持。

作为一种解决方法，您可以考虑将页脚附加到使用后续 ParDO 写入的所有文件（WriteToText 返回一个 PCollection 并写入所有文件）。

创建https://issues.apache.org/jira/browse/BEAM-10938 用于跟踪。

【讨论】：