Apache BeamでWRITE_TRUNCATEとWRITE_APPENDを動的にします

Question

GCSバケットに入っているファイルを処理し、一部のBigQueryテーブルにデータをダンプするApache Beamプログラムがあります。ファイルに応じて、切り捨てまたは追加操作を設定したいと思います。これは動的か設定可能にすることができますか？ ありがとうございます。

Answer 1

は、私はあなたが「ファイルに応じて、」を言うとき、あなたはファイルに関するいくつかの情報を持っていることを前提とし、あなたのパイプラインで（ときWRITE_TRUNCATEとWRITE_APPENDを使用するかを認識するために）。行うには

最も簡単な方法は、（フィルタリングによって）2 PCollectionsにあなたがBigQueryのに渡している入力を分割し、適切なBigQueryのシンク（WRITE_TRUNCATEとWRITE_APPENDと1と1）にそれらのそれぞれを渡すことになります。

は、JavaやPython、擬似コード使用する場合は言及しなかった以下のPythonのためであるが、それは簡単にJavaのSDKにご返信用

files = (pipeline 
     | 'Read files' >> beam.io.Read(Your_GCS_Source()) 
     ) 
files_to_truncate = (files 
     | beam.Filter(lambda file: filter_for_files_to_truncate()) 
     | beam.io.Write(beam.io.BigQuerySink(output_table, schema=output_schema, create_disposition=create_disposition, write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE)) 
    ) 

files_to_append = (files 
     | beam.Filter(lambda file: filter_for_files_to_append()) 
     | beam.io.Write(beam.io.BigQuerySink(output_table, schema=output_schema, create_disposition=create_disposition, write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)) 
    )