Insert data in BigQuery from Dataflow

Question

Previously, had PCollection formattedResults; and I was using below code to insert rows in big query:

                   // 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));

And all rows was directly inserted in BigQuery, all well till here. But now I have started to dynamically identify the table name and its row so am creating PCollection as below: (String will be the table name and then its row as value)

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

Also, I am creating group of rows which will go in same table as:

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

where key (String) is the BQ table name and value is the list of rows to be inserted in BQ.

With option 1, I was able to easily insert rows in BQ using code shown above but same code cannot be used with OPTION 2 or OPTION 3 because in this case my table name is key in map. Is there a way to insert rows in table using OPTION 2 or OPTION 3. Any link or code sample will be great help.

danielm danielm · Accepted Answer · 2016-08-04T20:29:49

The closest thing that Dataflow is writing to a table per window (and you can create your own BoundedWindow subclass and WindowFn to include whatever data you want in the window). To do this, use

to(SerializableFunction<BoundedWindow,String> tableSpecFunction)

on BigQueryIO.Write.

Note that this functionality uses BigQuery's streaming upload feature, which is limited to 100MB/s per table. Additionally, uploads are not atomic, so a failed batch job may upload only part of the output.

Insert data in BigQuery from Dataflow

2 Answers