From 0babad939b3167479119b367eaf53a694dfd7c5e Mon Sep 17 00:00:00 2001 From: mukeshbhandarkar Date: Fri, 4 Sep 2026 22:38:19 +0530 Subject: [PATCH] Document Python BigQuery CDC record format --- sdks/python/apache_beam/io/gcp/bigquery.py | 9 ++- .../io/built-in/google-bigquery.md | 73 +++++++++++++++++++ 2 files changed, 78 insertions(+), 4 deletions(-) diff --git a/sdks/python/apache_beam/io/gcp/bigquery.py b/sdks/python/apache_beam/io/gcp/bigquery.py index 40f17bfe9b0e..38acd29da7d9 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery.py +++ b/sdks/python/apache_beam/io/gcp/bigquery.py @@ -2241,10 +2241,11 @@ def __init__( max_insert_payload_size: The maximum byte size for a BigQuery legacy streaming insert payload. use_cdc_writes: Configure the usage of CDC writes on BigQuery. - The argument can be used by passing True and the Beam Rows will be - sent as they are to the BigQuery sink which expects a 'record' - and 'row_mutation_info' properties. - Used for STORAGE_WRITE_API, working on 'at least once' mode. + When True, requires ``method=STORAGE_WRITE_API`` and + ``use_at_least_once=True``. Each input element must contain ``record`` + and ``row_mutation_info`` properties. The latter must contain the + required string properties ``mutation_type`` (``UPSERT`` or + ``DELETE``) and ``change_sequence_number``. primary_key: When using CDC write on BigQuery and CREATE_IF_NEEDED mode for the underlying tables a list of column names is required to be configured as the primary key. Used for diff --git a/website/www/site/content/en/documentation/io/built-in/google-bigquery.md b/website/www/site/content/en/documentation/io/built-in/google-bigquery.md index 00cd1ffbcdef..418474a4b70f 100644 --- a/website/www/site/content/en/documentation/io/built-in/google-bigquery.md +++ b/website/www/site/content/en/documentation/io/built-in/google-bigquery.md @@ -904,6 +904,79 @@ When using `STORAGE_API_AT_LEAST_ONCE`, the `PCollection` returned by [`WriteResult.getFailedStorageApiInserts`](https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/io/gcp/bigquery/WriteResult.html#getFailedStorageApiInserts--) contains the rows that failed to be written to the Storage Write API sink. +#### Change data capture writes + + + +{{< paragraph class="language-py" >}} +The Python SDK supports BigQuery change data capture (CDC) writes with the +Storage Write API. To enable CDC writes, set `method` to +`WriteToBigQuery.Method.STORAGE_WRITE_API`, `use_at_least_once=True`, and +`use_cdc_writes=True`. +{{< /paragraph >}} + +{{< paragraph class="language-py" >}} +The destination table must have a primary key. BigQuery doesn't enforce primary +key uniqueness, so make sure that the key values are unique. When +`CREATE_IF_NEEDED` is used, pass the destination column names in the +`primary_key` argument so that Beam can create the table with that primary key. +For an existing table that already has a primary key, you can use +`CREATE_NEVER` without setting `primary_key`. +{{< /paragraph >}} + +{{< paragraph class="language-py" >}} +Each element in the input `PCollection` must contain a `row_mutation_info` Row +and a `record` Row. The `record` Row contains the destination table columns. +The `row_mutation_info` Row contains the required string fields `mutation_type` +and `change_sequence_number`. The mutation type must be `UPSERT` or `DELETE`. +{{< /paragraph >}} + +{{< highlight py >}} +mutations = [ + beam.Row( + row_mutation_info=beam.Row( + mutation_type="UPSERT", change_sequence_number="1"), + record=beam.Row(id=100, name="Alice")), + beam.Row( + row_mutation_info=beam.Row( + mutation_type="DELETE", change_sequence_number="2"), + record=beam.Row(id=100, name="Alice")), +] + +mutations = pipeline | beam.Create(mutations) + +mutations | beam.io.WriteToBigQuery( + table="project:dataset.table", + method=beam.io.WriteToBigQuery.Method.STORAGE_WRITE_API, + use_at_least_once=True, + use_cdc_writes=True, + primary_key=["id"]) +{{< /highlight >}} + +{{< paragraph class="language-py" >}} +Each element can instead be a dictionary whose `row_mutation_info` and `record` +values are nested dictionaries. For dictionary input, supply an explicit +compatible schema to `WriteToBigQuery`. Include `row_mutation_info` and `record` +in that schema, and make the `mutation_type` and `change_sequence_number` fields +required strings. +{{< /paragraph >}} + +{{< paragraph class="language-py" >}} +BigQuery uses `change_sequence_number` to order mutations that have the same +primary key. Sequence numbers contain one to four slash-separated hexadecimal +sections, with at most 16 hexadecimal digits in each section. Mutations with a +greater sequence number take precedence. +{{< /paragraph >}} + +{{< paragraph class="language-py" >}} +Beam translates `row_mutation_info.mutation_type` and +`row_mutation_info.change_sequence_number` to BigQuery's `_CHANGE_TYPE` and +`_CHANGE_SEQUENCE_NUMBER` metadata, respectively. Don't include those BigQuery +pseudo-fields directly in `record`. For complete BigQuery CDC semantics and +limitations, see the +[BigQuery CDC documentation](https://cloud.google.com/bigquery/docs/change-data-capture). +{{< /paragraph >}} + #### Tune the Storage Write API By default, the BigQueryIO Write transform uses Storage Write API settings that