Unable to run multiple UPDATE with BigQuery via Python SDK

userRRQl1 0
Viewed 111

I'm working on an ETL with Apache Beam and Dataflow using Python and I'm using BigQuery as a database/datawarehouse.

The ETL basically performs some processing then updates data that is already in BigQuery. Since there is no update transform in Apache Beam, I had to use the BigQuery SDK and write my own UPDATE query, and map it to each row.

The queries work fine when done sequentially, but when I use multiple workers, I get the following error:

{'reason': 'invalidQuery', 'message': 'Could not serialize access to table my_table due to concurrent update'}

I made sure that the same row is never accessed/updated concurrently (a row is basically an id, and each id is unique), I've also tried to run the same code with a simple Python script without Beam/Dataflow, and I still got the same error when I started using multiple threads instead of one.

Has anyone got the same problem using BigQuery SDK ? And do you have any suggestions to avoid that problem ?

apache-beam google-bigquery python
3 Answers

I think it's better from your Beam Dataflow job to append the data. Bigquery is more append oriented and the BigueryIO in Beam is adapted for append operation.

If you have an orchestrator like Cloud Composer/Airflow or Cloud Workflows, you can deduplicate the data in batch mode with the following steps :

  • Create a staging and final tables
  • Your orchestrator truncates your staging table
  • Your orchestrator runs your Dataflow job
  • Dataflow job reads your data
  • Dataflow job writes the result in append mode to Bigquery in the staging table
  • Your orchestrator run a task with a merge query with Bigquery between the staging and final tables. The merge query allows to insert or update the line in the final table if the element exists.

https://cloud.google.com/bigquery/docs/reference/standard-sql/dml-syntax?hl=en#merge_statement

Example of a merge query :

MERGE dataset.Inventory T
USING dataset.NewArrivals S
ON T.product = S.product
WHEN MATCHED THEN
  UPDATE SET quantity = T.quantity + S.quantity
WHEN NOT MATCHED THEN
  INSERT (product, quantity) VALUES(product, quantity)
Mazlum Tosun0

I had a use case where I had a BQ table that was containing around 150K records, and I needed to update its content monthly (which means around 100K UPDATE and couple of thousands APPEND.

When I designed my Beam/Dataflow job to update the records with the BQ python API library, I fall in Quota issues (limited number of updated) as well as the concurrency issue.

I had to change the approach my pipeline was working with, from reading the BQ table and updating the record, to process the BQ table, update what needs to be updated, and append what's new, and save to a new BQ table.

Once the job is successfully finished with no error, you can replace the old one with the new created table.

Idhem0

GCP mentions:

Running two mutating DML statements concurrently against a table will succeed as long as the two statements don’t modify data in the same partition. Two jobs that try to mutate the same partition may sometimes experience concurrent update failures.

And then :

BigQuery now handles such failures automatically. To do this, BigQuery will restart the job.

Can this retry mechanism be a solution at all ? Anyone to elaborate on this?

Source: https://cloud.google.com/blog/products/data-analytics/dml-without-limits-now-in-bigquery

Olivier D.0
Related