如何用Dataflow从GCS读取Avro文件为PCollection<TableRow>

首先你需要明确Avro文件对应的Java实体类（如果没有，可通过Avro工具根据Schema生成），然后用AvroIO.read()方法读取文件，再通过ParDo将Avro对象映射为TableRow。

完整代码示例 #

import org.apache.beam.sdk.io.AvroIO;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import com.google.api.services.bigquery.model.TableRow;

// 假设你的Avro数据对应的实体类是MyAvroRecord
public static PCollection<TableRow> avroFileReader(Pipeline pipeline, String inputAvroFile) {
    return pipeline.apply("Read Avro from GCS", 
            AvroIO.read(MyAvroRecord.class) // 指定Avro对应的实体类
                  .from(inputAvroFile)) // 传入GCS存储桶路径
        .apply("Convert Avro to TableRow", ParDo.of(new DoFn<MyAvroRecord, TableRow>() {
            @ProcessElement
            public void processElement(ProcessContext c) {
                MyAvroRecord record = c.element();
                TableRow row = new TableRow();
                // 根据Avro字段逐一映射到TableRow，示例：
                row.set("id", record.getId());
                row.set("user_name", record.getUserName());
                // 其他字段依次完成映射
                c.output(row);
            }
        }));
}

关键补充说明 #

• 若未预先定义Avro实体类，可使用AvroIO.readGenericRecords()读取通用记录后再转换：

// 传入Avro Schema读取通用记录的方式
pipeline.apply("Read Avro from GCS", 
        AvroIO.readGenericRecords(yourAvroSchema)
              .from(inputAvroFile))

• GCS路径支持通配符，比如gs://your-bucket/avro-data/*.avro可批量读取多个文件
• 运行Dataflow时需确保引入Avro相关依赖，Maven中需添加org.apache.beam:beam-sdks-java-io-avro

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