[Python] PCollection Objects Format for Apache Beam to write on BigQuery using CDC in Python

I'm trying to write to BigQuery using Apache Beam, in python. However, I want to use the newest CDC features to write on Bigquery.

However, I can't get the correct format of the objects in the PCollection. The documentation only says that, when using the "use_cdc_writes" option, "the Beam Rows will be sent as they are to the BigQuery sink which expects a ‘record’ and ‘row_mutation_info’ properties".

The objects in my Input PCollection are dictionaries with many fields. How can I format the objects as "Bean Rows" wirth "record" and "row_mutation_info" properties?

pcoll_output = (pcoll_input # | DataPointict[str,Any] |
| f'{table_name}:Write to BigQuery' >> WriteToBigQuery(
full_table_id, # Target table in BigQuery, e.g. `my-project.my_dataset.my_table_cdc`

# Table creation options
create_disposition=BigQueryDisposition.CREATE_NEVER, # The table should already exists
primary_key=None, #type:ignore Used only when creating the table. It should not be used here.

# Write Method Options
write_disposition=BigQueryDisposition.WRITE_APPEND, # Used for STREAMING
schema=table_config.schema.bigquery_schema_string, # Required for usage of STORAGE_WRITE_API
method=WriteToBigQuery.Method.STORAGE_WRITE_API, # MUCH CHEAPER THAN THE OTHER OPTIONS!

# CDC/Streaming Options
use_at_least_once=True, # True is cheaper and faster but might duplicate records.
use_cdc_writes=True, # Can only be True if use_at_least_once=True and method=STORAGE_WRITE_API
#ignore_insert_ids=False, # True is cheaper and faster but might duplicate records even when using CDC

# Test/Retry options
validate=True, # Validate the data in Apache Beam before writing to BigQuery. Use for testing
insert_retry_strategy=bigquery_tools.RetryStrategy.RETRY_ON_TRANSIENT_ERROR
)
)


I tried several ways of configuring the datapoints, as:

• Including a "row_mutation_info" field with the mutation info
• Including a "record" field with the remaining data of each Datapoint
• Formating both as such:

datapoint = Row(
row_mutation_info=Row(
mutation_type=self.compute_change_type(datapoint),
change_sequence_number=self.format_sequence_number(datapoint)
),
record=bigquery_tools.beam_row_from_dict(datapoint, self.schema)
)

Continue reading...