BigQueryIO.Writeの前にCSVヘッダーを読み込んで変換するにはどうすればよいですか？

Question

私は最初の行としてヘッダーを含むcsvファイルを持っています。私はそれを読んで、BigQueryのカラム要件に合うようにこれらのヘッダを整理しています。しかし、パイプラインが始まるの前にスキーマへの参照が必要です。 BigQueryIO.Writeがこのようにヘッダーに応答できるようにするためのベストプラクティスは何ですか？現在、私のコードは次のようなものになります。私は現在、次のように（おおよそ）見て、おそらく2つのパイプラインを実装しようとしている

//create table 
Table table = new Table(); 
// Where logically should the following line go? 
TableSchema customSchema = ? 
table.setSchema(customSchema); 
TableReference tableRef = new TableReference(); 
tableRef.setDatasetId("foo_dataset"); 
tableRef.setProjectId("bar_project"); 
tableRef.setTableId("baz_table"); 
table.setTableReference(tableRef); 

Pipeline p = Pipeline.create(options); 

p.apply(TextIO.Read.named("ReadCSV").from("gs://bucket/file.csv")) 
    // Detect if it's header row 
    .apply(ParDo.of(new ExtractHeader())) 
    .apply(ParDo.of(new ToTableRow()) 
    .apply(BigQueryIO.Write.named("Write") 
    .to(tableRef) 
    // Where logically should the following line go? 
    .withSchema(customSchema)); 
p.run(); 

を、しかし、実行順序は、データフローに私が取得していますエラーがどこBQ信頼できませんテーブルは存在しません。

PCollection readIn = p.apply(TextIO.Read.named("ReadCSV").from("gs://bucket/file.csv")) 
    .apply(ParDo.of(new ExtractHeader())); 
TableSchema customSchema = /* generate schema based on what I now know the headers are */ 
readIn.apply(ParDo.of(new ToTableRow()) 
    .apply(BigQueryIO.Write.named("Write") 
    .to(tableRef) 
    // Where logically should the following line go? 
    .withSchema(customSchema)); 
p.run(); 

Answer 1

この機能（動的スキーマ）は現在レビュー中ですhttps://github.com/apache/beam/pull/2609（私はそれを検討しています）。進行中のPRを試すことができますが、そのAPIはレビューの結果、多少変更される可能性があることに注意してください。 PRが提出されたら、この回答を更新します。