Apache Beam读取BigQuery为Avro格式Generic Records的实现方案

存在原生支持的实现方式，无需经过TableRow转换环节，具体操作如下：

Java SDK 实现方式 #

使用BigQueryIO.TypedRead接口直接指定返回类型为GenericRecord，底层会直接将BigQuery存储层的Avro格式数据映射为对应实例，不需要TableRow中转：

// 1. 定义与BigQuery表结构匹配的Avro Schema，支持手动编写或调用BigQuery API自动生成
Schema tableAvroSchema = ...;

PCollection<GenericRecord> bqRecords = pipeline.apply("读取BigQuery为GenericRecord",
  BigQueryIO.read(schemaAndRecord -> schemaAndRecord.getRecord())
    .from("your-project-id:your-dataset.your-table-name")
    .withAvroSchema(tableAvroSchema)
    // 开启直读模式，跳过导出到GCS的中间环节，性能更好
    .withMethod(BigQueryIO.TypedRead.Method.DIRECT_READ)
);

Python SDK 实现方式 #

读取时指定输出格式为AVRO，返回结果直接为GenericRecord类型：

import apache_beam as beam
from apache_beam.io.gcp.bigquery import BigQuerySource

avro_schema = {
    # 对应BigQuery表结构的Avro Schema定义
}

with beam.Pipeline() as p:
  records = p | beam.io.Read(BigQuerySource(
      table='your-project-id:your-dataset.your-table-name',
      output_type='AVRO',
      avro_schema=avro_schema
  ))

注意事项 #

• 传入的Avro Schema字段顺序、类型需要和BigQuery表结构完全一致，否则会触发类型映射错误
• 直读模式对大表的读取性能优于默认的导出读取模式，适合需要低延迟读取的场景

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