Формат объектов Pcollection для Apache Beam для записи на BigQuery с использованием CDC в Python ⇐ Python

[Anonymous]

Я пытаюсь написать в BigQuery, используя Apache Beam, в Python.
Однако я хочу использовать новейшие функции CDC для написания на BigQuery. Ожидает свойства «записи» и «ROW_MUTICE_INFO». Как я могу отформатировать объекты как свойства "Bean Rows" Wirth "и" и "ROW_MUTICE_INFO"? < /P>
pcoll_output = (pcoll_input # | DataPoint:Dict[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
)
)
< /code>
Я попробовал несколько способов настройки данных, как: < /p>

включает в себя полю "row_mute_info" с информацией о мутации < /li>
, как «запись» с оставшимися данными каждого DataPoint < /li>
< /li> < /li> < /li> < /li> < /li> < /li> < /li> < /li> < /li> < /li> < /li> < /li> < /li> < /li> < /li> < /li> < /li> < /br /> /> < /ul>
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)
)


Подробнее здесь: https://stackoverflow.com/questions/796 ... cdc-in-pyt