Googleのデータフロー入力に基づいて複数のテーブルに書き込む

Question

Google BigQueryにプッシュしようとしているログがあります。私はGoogleのデータフローを使用してパイプライン全体を構築しようとしています。ログ構造は異なり、4つの異なるタイプに分類できます。私のパイプラインでは、PubSubからログを読み込み、それを解析してBigQueryテーブルに書き込みます。ログが書き込まれる必要があるテーブルは、ログの1つのパラメータに依存します。問題は、実行時にBigQueryIO.WriteのTableNameを変更する方法のポイントに固執しています。

Answer 1

サイド出力を使用できます。

https://cloud.google.com/dataflow/model/par-do#emitting-to-side-outputs-in-your-dofn

以下のサンプルコードは、BigQueryのテーブルを読み出して、3つの異なるPCollectionsでそれを分割します。それぞれのPCollectionsは別のPub/Subトピック（代わりに別のBigQueryテーブル）に送信されてしまいます。

Pipeline p = Pipeline.create(PipelineOptionsFactory.fromArgs(args).withValidation().create()); 

PCollection<TableRow> weatherData = p.apply(
     BigQueryIO.Read.named("ReadWeatherStations").from("clouddataflow-readonly:samples.weather_stations")); 

final TupleTag<String> readings2010 = new TupleTag<String>() { 
}; 
final TupleTag<String> readings2000plus = new TupleTag<String>() { 
}; 
final TupleTag<String> readingsOld = new TupleTag<String>() { 
}; 

PCollectionTuple collectionTuple = weatherData.apply(ParDo.named("tablerow2string") 
     .withOutputTags(readings2010, TupleTagList.of(readings2000plus).and(readingsOld)) 
     .of(new DoFn<TableRow, String>() { 
      @Override 
      public void processElement(DoFn<TableRow, String>.ProcessContext c) throws Exception { 

       if (c.element().getF().get(2).getV().equals("2010")) { 
        c.output(c.element().toString()); 
       } else if (Integer.parseInt(c.element().getF().get(2).getV().toString()) > 2000) { 
        c.sideOutput(readings2000plus, c.element().toString()); 
       } else { 
        c.sideOutput(readingsOld, c.element().toString()); 
       } 

      } 
     })); 
collectionTuple.get(readings2010) 
     .apply(PubsubIO.Write.named("WriteToPubsub1").topic("projects/fh-dataflow/topics/bq2pubsub-topic1")); 
collectionTuple.get(readings2000plus) 
     .apply(PubsubIO.Write.named("WriteToPubsub2").topic("projects/fh-dataflow/topics/bq2pubsub-topic2")); 
collectionTuple.get(readingsOld) 
     .apply(PubsubIO.Write.named("WriteToPubsub3").topic("projects/fh-dataflow/topics/bq2pubsub-topic3")); 

p.run();