如何为 apache 光束数据流的输出 csv 添加标头？答案

【问题标题】：How do I add headers for the output csv for apache beam dataflow?如何为 apache 光束数据流的输出 csv 添加标头？
【发布时间】：2016-09-21 19:15:07
【问题描述】：

我注意到在 java sdk 中有一个函数可以让您编写 csv 文件的标题。 https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/io/TextIO.Write.html#withHeader-java.lang.String-

此功能是否反映在 python skd 上？

【问题讨论】：

标签： python google-cloud-dataflow apache-beam

【解决方案1】：

您现在可以使用文本接收器写入文本并指定标题。

来自文档：

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)

所以你可以使用下面这段代码：

beam.io.WriteToText(bucket_name, file_name_suffix='.csv', header='colname1, colname2')

如果您想了解详细信息或查看其实现方式，请在此处获得完整的文档：https://beam.apache.org/documentation/sdks/pydoc/2.0.0/_modules/apache_beam/io/textio.html#WriteToText

【讨论】：

【解决方案2】：

目前尚未实施。但是，您可以自己实现/扩展它（请参阅attached notebook 以获取我的 apache_beam 版本的示例+测试）。

这是基于超类FileSink 的note in the docstring，提到你应该覆盖open 函数：

适用于我的 apache_beam 版本的新类（'0.3.0-incubating.dev'）：

import apache_beam as beam
from apache_beam.io import TextFileSink
from apache_beam.io.fileio import ChannelFactory,CompressionTypes
from apache_beam import coders


class TextFileSinkWithHeader(TextFileSink):
    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.NO_COMPRESSION,
               header=None):
        super(TextFileSinkWithHeader, self).__init__(
            file_path_prefix,
            file_name_suffix=file_name_suffix,
            num_shards=num_shards,
            shard_name_template=shard_name_template,
            coder=coder,

            compression_type=compression_type,
            append_trailing_newlines=append_trailing_newlines)
        self.header = header

    def open(self, temp_path):
        channel_factory = ChannelFactory.open(
            temp_path,
            'wb',
            mime_type=self.mime_type)
        channel_factory.write(self.header+"\n")
        return channel_factory

您随后可以按如下方式使用它：

beam.io.Write(TextFileSinkWithHeader('./names_w_headers',header="names"))

有关完整概述，请参阅 the notebook。

【讨论】：

【解决方案3】：

Python SDK 中尚不存在此功能

【讨论】：

【解决方案4】：

对于 Python SDK：

beam.io.Write(beam.io.WriteToText(
                     file_path_prefix=os.path.join(OUTPUT_DIR),
                     file_name_suffix='.csv', header='colname1, colname2')
              )

【讨论】：