从数据流管道写入BQ时的动态表名

Question

作为以下问答的后续问题：

https://stackoverflow.com/questions/31156774/about-key-grouping-with-groupbykey

我想与谷歌数据流工程团队(@jkff)确认尤金提出的第三个选项是否有可能使用谷歌数据流：

PCollection<KV<String, Iterable<String>>> output = ...;

public class DynamicOutput2Fn extends DoFn<KV<String, Iterable<String>>, Integer> {

private final PCollectionView<List<String>> keysAsSideinputs;
public DynamicOutput2Fn(PCollectionView<List<String>> keysAsSideinputs) {
        this.keysAsSideinputs = keysAsSideinputs;
    }

@Override
    public void processElement(ProcessContext c) {
        List<String> keys = c.sideInput(keysAsSideinputs);
        String key = c.element().getKey();

        //the below is not working!!! How could we write the value out to a sink, be it gcs file or bq table???
        c.element().getValue().apply(Pardo.of(new FormatLineFn()))
                .apply(TextIO.Write.to(key));

        c.output(1);
    }    
}    

Answer 1

BigQueryIo.Write转换不支持这一点。您可以做的最接近的事情是使用每个窗口的表，并使用自定义WindowFN对选择窗口对象中的表所需的任何信息进行编码。

如果不想这样做，可以直接从DOFN调用BigQuery API。这样，您就可以将表名设置为任何您想要的名称，正如您的代码所计算的那样。这可以从侧输入中查找，也可以直接从DoFn当前正在处理的元素中计算。为了避免对BigQuery进行太多的小调用，可以使用finishBundle（）对请求进行批量处理；

您可以在这里看到数据流运行器是如何导入流的：https://github.com/googlecloudplatform/dataflowjavasdk/blob/master/sdk/src/main/java/com/google/cloud/Dataflow/sdk/util/bigquerytableinserter.java