云数据流、PubSub和Bigquery问题

Question

我想使用Cloud Dataflow,PubSub和Bigquery将tableRow写入PubSub消息,然后将它们写入Bigquery。我希望表名、项目id和数据集id是动态的。
我在internet上看到下面的代码，我不明白如何传递数据行参数。

public void PubSub(String projectId , String datasetId,String tableId,String topicId)       
    PipelineOptions options = PipelineOptionsFactory.create();
    DataflowPipelineOptions dataflowOptions = options.as(DataflowPipelineOptions.class);
    dataflowOptions.setStreaming(true);
    Pipeline pipeline = Pipeline.create(dataflowOptions);
    PCollection<TableRow> input = pipeline.apply(PubsubIO.Read.topic(createTopic(projectId,topicId).getName()).withCoder(TableRowJsonCoder.of()))
            .apply(Window.<TableRow>into(FixedWindows.of(Duration.standardMinutes(1))));

    input.apply(BigQueryIO.Write.to(getTableReference(projectId,datasetId, tableId)).withSchema(getSchema()));

    pipeline.run();
}


private static TableReference getTableReference(String projectId , String datasetId,String tableId) {
      TableReference tableRef = new TableReference();
      tableRef.setProjectId(projectId);
      tableRef.setDatasetId(datasetId);
      tableRef.setTableId(tableId);
      return tableRef;
}

先谢谢你，盖尔

Answer 1

BigQueryIo.Write转换不支持动态输出。但是您可以直接从DOFN进行BigQuery API调用。

这样，您就可以根据代码的计算，将表名设置为任何您想要的名称。这可以从侧面输入中查找，或者直接从DoFn当前正在处理的元素中计算。

为了避免对BigQuery进行太多的小调用,可以使用finishBundle（）对请求进行批处理；

我不完全理解您是否要将Dataflow写到pub/sub，然后将pub/sub写到bigquery？您可以直接写入BigQuery而不使用pub/sub。