如何将Dataset<Row>转换为List<GenericRecord>？

针对从Iceberg表（底层以Avro格式存储）读取的Dataset<Row>，要转换为List<GenericRecord>，这里提供两种高效实现方案，同时先说明你当前思路的问题：

你当前思路的问题 #

直接将Row转成byte[]再用DataFileReader反序列化的方式不可行——DataFileReader是用于读取完整Avro数据文件的（包含文件头、元数据等固定结构），而单个Row的字节流并非标准Avro数据文件格式，会直接导致反序列化失败。此外这种方式存在冗余，Dataset已经是解析后的结构化数据，直接基于Schema映射更高效。

方案一：利用Spark Avro库直接转换 #

Spark官方提供了Avro扩展库，可以快速将Dataset<Row>序列化为Avro字节，再反序列化为GenericRecord，能严格保证Schema一致性。

步骤1：引入依赖

确保项目中包含对应Spark版本的Avro依赖：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-avro_2.12</artifactId>
    <version>3.5.0</version> <!-- 替换为你的Spark版本 -->
</dependency>

步骤2：代码实现

import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.DatumReader;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.avro.AvroFunctions;
import org.apache.spark.sql.Encoders;

// 从Iceberg读取的Dataset<Row>
Dataset<Row> data = spark.sql(SQL_QUERY);

// 将Dataset转换为包含Avro字节的Dataset<byte[]>
Dataset<byte[]> avroBytesDs = data.select(AvroFunctions.to_avro(data.schema()).alias("avro_bytes"));

// 收集并反序列化为GenericRecord列表
List<GenericRecord> genericRecords = avroBytesDs.map(row -> {
    byte[] bytes = row.getAs("avro_bytes");
    DatumReader<GenericRecord> reader = new GenericDatumReader<>();
    return reader.read(null, DecoderFactory.get().binaryDecoder(bytes, null));
}, Encoders.kryo(GenericRecord.class)).collectAsList();

方案二：手动映射Row到GenericRecord（自定义场景） #

如果需要对字段进行自定义转换或特殊处理，可以手动基于Schema映射Row到GenericRecord。

代码实现

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.avro.SchemaConverters;
import org.apache.spark.sql.Encoders;

// 从Iceberg读取的Dataset<Row>
Dataset<Row> data = spark.sql(SQL_QUERY);
StructType sparkSchema = data.schema();

// 将Spark Schema转换为对应的Avro Schema（也可直接从Iceberg表元数据获取）
Schema avroSchema = SchemaConverters.toAvroType(sparkSchema);

// 逐行转换为GenericRecord
List<GenericRecord> genericRecords = data.map(row -> {
    GenericRecord record = new GenericData.Record(avroSchema);
    for (int i = 0; i < sparkSchema.fields().length; i++) {
        String fieldName = sparkSchema.fields()[i].name();
        Object value = row.get(i);
        // 复杂类型（如数组、结构体）可在此处添加自定义转换逻辑
        record.put(fieldName, value);
    }
    return record;
}, Encoders.kryo(GenericRecord.class)).collectAsList();

内容的提问来源于stack exchange，提问作者Roni Koren Kurtberg