Write to a dynamic BigQuery table through Apache Beam

Question

I am getting the BigQuery table name at runtime and I pass that name to the BigQueryIO.write operation at the end of my pipeline to write to that table.

The code that I've written for it is:

rows.apply("write to BigQuery", BigQueryIO
    .writeTableRows()
    .withSchema(schema)
    .to("projectID:DatasetID."+tablename)
    .withWriteDisposition(WriteDisposition.WRITE_TRUNCATE)
    .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED));

With this syntax I always get an error,

Exception in thread "main" java.lang.IllegalArgumentException: Table reference is not in [project_id]:[dataset_id].[table_id] format

How to pass the table name with the correct format when I don't know before hand which table it should put the data in? Any suggestions?

Thank You

How many table can be there to write the data into bigquery? — Manoj Kumar
Where is the "tableName" field or variable coming from? How is it defined? — Ben Chambers
Also: this error message seems incomplete; I believe it should include the actual value of the table spec. It is difficult to help you with this error without knowing the exact value being passed to .to() - the error message simply says that the value you passed to .to() is malformed; it is not related to the fact that you are passing it at runtime. — jkff
Perhaps another thing you'll find useful: see this answer stackoverflow.com/a/43505535/278042 if the tables you're writing to are determined by the data itself. — jkff
@BenChambers What I'm doing is, I'm getting the "tableName" from a field in a BigQuery table and storing it in a String variable...which I'm ultimately passing to the BigQueryIO.write() operation. — rish0097

Campey Campey · Accepted Answer · 2018-05-28T06:34:28

Very late to the party on this however. I suspect the issue is you were passing in a string not a table reference.

If you created a table reference I suspect you'd have no issues with the above code.

com.google.api.services.bigquery.model.TableReference table = new TableReference()
            .setProjectId(projectID)
            .setDatasetId(DatasetID)
            .setTableId(tablename);

rows.apply("write to BigQuery", BigQueryIO
    .writeTableRows()
    .withSchema(schema)
    .to(table)
    .withWriteDisposition(WriteDisposition.WRITE_TRUNCATE)
    .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED));

Write to a dynamic BigQuery table through Apache Beam

1 Answers