使用AvroParquetWriter写入POJO列表时抛出ArrayIndexOutOfBoundsException

你定义的Avro Schema是数组类型（直接返回SchemaBuilder.array().items(...)），但实际通过writer.write()传入的是单个ServiceOrderJson对应的GenericRecord，两者结构不匹配。AvroParquetWriter期望写入的是一个数组对象，但实际收到的是单个Record，导致Parquet在解析字段索引时出现越界错误。

根据你的需求（保存POJO对象列表），有两种可行的修正方式：

方案1：使用单个Record Schema，循环写入每个对象 #

这是最常见的场景：定义单个ServiceOrderJson的Record Schema，然后遍历POJO列表，逐个写入对应的GenericRecord。

修正Schema定义 #

public static Schema getServiceOrderJsonSchema() {
    return SchemaBuilder.record("ServiceOrderJson")
            .namespace("ServiceOrders")
            .fields()
                .name("value1").type().stringType().noDefault()
                .name("value2").type().stringType().noDefault()
                .name("value3").type().longType().noDefault()
                .name("value4").type().stringType().noDefault()
                .name("value5").type().booleanType().noDefault()
            .endRecord();
}

修正写入逻辑 #

确保遍历POJO列表，逐个转换并写入：

// 假设serviceOrders是List<ServiceOrderJson>
ParquetWriter<GenericRecord> writer = AvroParquetWriter
        .<GenericRecord>builder(path)
        .withSchema(getServiceOrderJsonSchema())
        .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
        .withConf(config)
        .withValidation(false)
        .withPageWriteChecksumEnabled(false)
        .withDictionaryEncoding(false)
        .withCompressionCodec(CompressionCodecName.SNAPPY)
        .build();

for (ServiceOrderJson item : serviceOrders) {
    writer.write(convertToGenericRecord(item, getServiceOrderJsonSchema()));
}
writer.close();

方案2：定义包含数组字段的根Record（适用于需要根为数组的场景） #

如果必须让Parquet文件的根结构是数组，可以定义一个包含数组字段的根Record，将所有POJO对象放入数组后一次性写入。

修正Schema定义 #

public static Schema getServiceOrdersRootSchema() {
    // 先定义单个ServiceOrderJson的Schema
    Schema itemSchema = SchemaBuilder.record("ServiceOrderJson")
            .namespace("ServiceOrders")
            .fields()
                .name("value1").type().stringType().noDefault()
                .name("value2").type().stringType().noDefault()
                .name("value3").type().longType().noDefault()
                .name("value4").type().stringType().noDefault()
                .name("value5").type().booleanType().noDefault()
            .endRecord();
    
    // 定义根Record，包含一个数组字段
    return SchemaBuilder.record("ServiceOrdersRoot")
            .namespace("ServiceOrders")
            .fields()
                .name("orders").type().array().items(itemSchema).noDefault()
            .endRecord();
}

修正写入逻辑 #

构造包含数组的根GenericRecord后写入：

// 将所有POJO转换为GenericRecord列表
List<GenericRecord> recordList = new ArrayList<>();
for (ServiceOrderJson item : serviceOrders) {
    recordList.add(convertToGenericRecord(item, getServiceOrdersRootSchema().getField("orders").schema().getElementType()));
}

// 构造根Record
GenericRecord rootRecord = new GenericData.Record(getServiceOrdersRootSchema());
rootRecord.put("orders", recordList);

// 写入根Record
ParquetWriter<GenericRecord> writer = AvroParquetWriter
        .<GenericRecord>builder(path)
        .withSchema(getServiceOrdersRootSchema())
        .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
        .withConf(config)
        .withValidation(false)
        .withPageWriteChecksumEnabled(false)
        .withDictionaryEncoding(false)
        .withCompressionCodec(CompressionCodecName.SNAPPY)
        .build();

writer.write(rootRecord);
writer.close();

• 确保convertToGenericRecord方法生成的GenericRecord字段顺序、类型与Schema完全匹配，字段名大小写一致。
• 若使用方案1，Parquet文件会将每个POJO作为独立的行存储，这更符合Parquet的列式存储优化特性，推荐优先使用。

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