Beam中是否支持将BigQuery TableRow转换为Avro GenericRecord？

Apache Beam 内置了对应的转换工具类，无需从零实现完整转换逻辑，具体使用方式如下：

• 核心依赖工具类为org.apache.beam.sdk.io.gcp.bigquery.BigQueryAvroUtils，该类封装了BigQuery Schema到Avro Schema转换、TableRow到GenericRecord转换的全链路能力。
• 单个对象转换步骤：
  1. 先获取BigQuery表对应的TableSchema实例，可以通过BigQuery元数据接口查询获取，也可以从已有的Schema JSON字符串反序列化得到
  2. 调用BigQueryAvroUtils.toGenericAvroSchema()方法，传入自定义的Avro记录名、BigQuery TableSchema的字段列表，生成对应的Avro Schema对象
  3. 调用BigQueryAvroUtils.convertBigQueryTableRowToGenericRecord()方法，传入待转换的TableRow对象和上一步生成的Avro Schema，直接得到转换完成的GenericRecord实例
• 单对象转换代码示例：

// 从JSON字符串反序列化得到BigQuery TableSchema
TableSchema bqSchema = BigQueryHelpers.fromJsonString(bqSchemaJsonStr, TableSchema.class);
// 生成对应Avro Schema
Schema avroSchema = BigQueryAvroUtils.toGenericAvroSchema("my_avro_record", bqSchema.getFields());
// TableRow转GenericRecord
GenericRecord avroRecord = BigQueryAvroUtils.convertBigQueryTableRowToGenericRecord(inputTableRow, avroSchema);

• 流水线批量转换示例：
  如果需要在Dataflow流水线中对PCollection<TableRow>做批量转换，可以封装为如下DoFn：

PCollection<TableRow> inputTableRows = ...;
PCollection<GenericRecord> outputAvroRecords = inputTableRows.apply(ParDo.of(new DoFn<TableRow, GenericRecord>() {
    private transient Schema avroSchema;

    @Setup
    public void initSchema() {
        // 预获取BigQuery表Schema，建议提前传入或在Setup阶段一次性加载
        TableSchema bqSchema = BigQueryOptions.newBuilder().build().getService()
            .tables().get(projectId, datasetId, tableId).execute().getSchema();
        avroSchema = BigQueryAvroUtils.toGenericAvroSchema("batch_output_record", bqSchema.getFields());
    }

    @ProcessElement
    public void convert(@Element TableRow row, OutputReceiver<GenericRecord> out) {
        out.output(BigQueryAvroUtils.convertBigQueryTableRowToGenericRecord(row, avroSchema));
    }
}));

• 注意事项：
  • 内置转换逻辑默认兼容所有BigQuery原生数据类型，包括嵌套RECORD、REPEATED数组、TIMESTAMP、DATE、BYTES等特殊类型，无需额外做类型映射
  • 如果有自定义业务字段转换需求，可以在调用转换方法前先对TableRow的字段值做预处理，再传入转换方法

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