Dataflow写入BigQuery存储Protobuf二进制数据时遇Unicode解码错误

这个错误的核心原因是Dataflow的BigQuery IO在自动推断字段类型时，误将你的二进制bytes数据当成了UTF-8字符串，尝试解码时失败——毕竟Protobuf序列化后的二进制数据本身就不是合法的UTF-8编码，自然会抛出UnicodeDecodeError。而BigQuery UI能正常插入，是因为UI直接识别了这是二进制数据，没有做额外的解码操作。

下面是两种可行的解决方法：

方法一：显式指定BigQuery表Schema #

最直接的方式是告诉Dataflow你的表字段类型是BYTES，避免它自动推断出错。

首先定义表的Schema：

from apache_beam.io.gcp.bigquery import TableSchema, TableFieldSchema

# 定义备份表的Schema，明确data字段为BYTES类型
backup_table_schema = TableSchema(
    fields=[
        TableFieldSchema(
            name='data',
            type='BYTES',
            mode='REQUIRED'
        )
    ]
)

然后修改WriteToBigQuery的调用，指定这个Schema：

bytes_status | 'Write to BQ BackUp' >> beam.io.WriteToBigQuery(
    'my_project:my_dataset.my_table',
    schema=backup_table_schema,
    # 根据你的需求选择写入策略，这里用追加
    write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
    # 如果表已经存在，设置为CREATE_NEVER避免重复创建
    create_disposition=beam.io.BigQueryDisposition.CREATE_NEVER
)

方法二：将二进制数据包装为Base64字符串（备选） #

如果你不想指定Schema，也可以把二进制数据转换成Base64编码的字符串，BigQuery会自动将Base64字符串解析为BYTES类型（前提是你的表字段确实是BYTES类型）。修改你的GetBytes DoFn：

import base64

class GetBytes(beam.DoFn):
  def process(self, element):
    # 将bytes转换为Base64字符串
    obj: Dict = { 'data': base64.b64encode(element.data).decode('utf-8') }
    logging.info(f'data bytes (base64): {obj}')
    return [obj]

这种方式不需要额外指定Schema，因为Base64字符串是合法的UTF-8，Dataflow会把它当成字符串传递给BigQuery，而BigQuery的BYTES字段会自动解码Base64字符串为二进制数据。

验证建议 #

优先选择方法一，因为它更直接匹配你的表结构，避免额外的编码解码开销。修改后重新运行管道，应该就能避免UnicodeDecodeError了。

内容的提问来源于stack exchange，提问作者Kimor