python - Apache Beam データフローの出力 csv のヘッダーを追加するにはどうすればよいですか?

Question

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 に反映されていますか?

Answer 1

これは現時点では実装されていません。ただし、自分で実装/拡張することはできます (私のバージョンの apache_beam での例とテストについては、添付のノートブックを参照してください)。

これは、スーパークラスの docstring のメモに基づいており、関数FileSinkを上書きする必要があると述べています。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"))

完全な概要については、ノートブックを参照してください。

Answer 2

この機能は Python SDK にはまだ存在しません