从 Dataflow 在 BigQuery 中插入数据答案

【问题标题】：Insert data in BigQuery from Dataflow从 Dataflow 在 BigQuery 中插入数据
【发布时间】：2016-08-04 18:54:53
【问题描述】：

以前，有 PCollection 格式化结果；我正在使用下面的代码在大查询中插入行：

                   // OPTION 1
PCollection<TableRow> formattedResults = ....
formattedResults.apply(BigQueryIO.Write.named("Write").to(tableName)
                            .withSchema(tableSchema)
                            .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
                            .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));

所有行都直接插入到 BigQuery 中，直到这里。但现在我已经开始动态识别表名及其行，所以创建 PCollection 如下：（字符串将是表名，然后将其行作为值）

PCollection<KV<String, TableRow>>   tableRowMap // OPTION 2

另外，我正在创建一组行，它们将在同一个表中：

PCollection<KV<String, Iterable<TableRow>>> groupedRows  //OPTION 3

其中 key (String) 是 BQ 表名，value 是要插入 BQ 中的行列表。

使用选项 1，我可以使用上面显示的代码轻松地在 BQ 中插入行，但相同的代码不能与 OPTION 2 或 OPTION 3 一起使用，因为在这种情况下，我的表名是 map 中的键。有没有办法使用 OPTION 2 或 OPTION 3 在表中插入行。任何链接或代码示例都会有很大帮助。

【问题讨论】：

标签： google-bigquery google-cloud-dataflow dataflow

【解决方案1】：

Dataflow 最接近每个窗口写入表的内容（您可以创建自己的 BoundedWindow 子类和 WindowFn 以在窗口中包含您想要的任何数据）。为此，请使用

to(SerializableFunction<BoundedWindow,String> tableSpecFunction)

在 BigQueryIO.Write 上。

请注意，此功能使用 BigQuery 的流式上传功能，每个表限制为 100MB/s。此外，上传不是原子的，因此失败的批处理作业可能只上传部分输出。

【讨论】：

【解决方案2】：

您还可以选择创建自己的 DoFn，将数据直接插入 bigquery，而不是依赖 BigQueryIO.Write。从技术上讲，您需要创建一个BigQueryTableInserter，您可以使用insertAll(TableReference ref, List<TableRow> rowList) 将内容插入到您想要的表中。

您可以使用以下方式创建 TableReference： new TableReference().setProjectId("projectfoo").setDatasetId("datasetfoo").setTableId("tablefoo")

这不是 100% 推荐的，因为 BigQueryIO 做了一些很好的事情来分割需要插入的行以最大限度地提高吞吐量并正确处理重试。

【讨论】：