如何为apache beam数据流的输出csv添加标题？

2

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

从文档：

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

来源

2017-06-27 17:04:18 nlassaux

0

此功能在Python SDK中尚不存在

来源

2016-09-21 19:57:26 danielm

1

此处未实现。但是你可以自己实现/扩展它（参见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的完整概述。

来源

2016-09-21 19:58:53 Fematich

如何为apache beam数据流的输出csv添加标题？

回答

相关问题