我注意到在java sdk中,有一個函數允許你編寫一個csv文件的頭文件。 https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/io/TextIO.Write.html#withHeader-java.lang.String-如何爲apache beam數據流的輸出csv添加標題?
此功能在python skd上鏡像嗎?
我注意到在java sdk中,有一個函數允許你編寫一個csv文件的頭文件。 https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/io/TextIO.Write.html#withHeader-java.lang.String-如何爲apache beam數據流的輸出csv添加標題?
此功能在python skd上鏡像嗎?
您現在可以寫入文本並使用文本接收器指定標題。
從文檔:
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
此功能在Python SDK中尚不存在
此處未實現。但是你可以自己實現/擴展它(參見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的完整概述。