Avro序列化、反序列化詳解及自定義工具類

1、前言

Avro序列化的API主要有兩種，SpecificDatumWriter / SpecificDatumReader及DataFileWriter / DataFileReader，后者是對前者的封裝。兩者的特點分別介紹如下：

2、SpecificDatumWriter / SpecificDatumReader

2.1 SpecificDatumWriter序列化

SpecificDatumWriter序列化一條或多條記錄

public static ByteArrayOutputStream serializePrimary(Schema schema, List<GenericRecord> records) throws IOException{
    DatumWriter<GenericRecord> datumWriter = new SpecificDatumWriter<GenericRecord>(schema);
    ByteArrayOutputStream out = new ByteArrayOutputStream();
    BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out , null);

    //多次調用write方法
    for(GenericRecord record : records){
        datumWriter.write(record , encoder);
        encoder.flush();
    }
    return out;
}

2.2 SpecificDatumReader反序列化

SpecificDatumReader反序列化獲得一條或多條記錄

public static List<GenericRecord> deserializeMulPrimary(Schema schema, ByteArrayOutputStream out) throws IOException{
    DatumReader<GenericRecord> datumReader = new SpecificDatumReader<GenericRecord>(schema);
    Decoder decoder=DecoderFactory.get().binaryDecoder(out.toByteArray(), null);
        
    List<GenericRecord> records = new ArrayList<GenericRecord>();
    while(true){
        try {
            GenericRecord record = datumReader.read(null, decoder);
            records.add(record);
        } catch (EOFException eof) {
            //讀取到字節流的末尾時，結束循環
            break;
        }
    }
    return records;
}

2.3 特點

a 序列化后的內容中不含有schema信息
b 反序列化時必須有schema信息（因為序列化記錄中沒有schema信息）
c 主要以內存為存儲媒介
d 可以序列化和反序列化獲得一條或多條記錄

3、 DataFileWriter / DataFileReader

3.1 DataFileWriter序列化

將數據序列化到內存中

public static ByteArrayOutputStream serializeToMemory(Schema schema, List<GenericRecord> records) throws IOException{
    DatumWriter<GenericRecord> datumWriter = new SpecificDatumWriter<GenericRecord>(schema);
    DataFileWriter<GenericRecord> fileWriter = new DataFileWriter<GenericRecord>(datumWriter);
        
    ByteArrayOutputStream out = new ByteArrayOutputStream();
    //先現將Schema寫入到內存中
    fileWriter.create(schema, out);
    //再開始追加多條GenericRecord記錄
    for(GenericRecord record : records){
        fileWriter.append(record);
    }
    fileWriter.close();
    return out;
}

將數據序列化到avro文件中

public static File serializeToFile(Schema schema, List<GenericRecord> records, String fileDirectory) throws IOException{
    DatumWriter<GenericRecord> datumWriter = new SpecificDatumWriter<GenericRecord>(schema);
    DataFileWriter<GenericRecord> fileWriter = new DataFileWriter<GenericRecord>(datumWriter);
        
    File file = new File(fileDirectory + "/" + System.currentTimeMillis() + "-" + UUID.randomUUID().toString() + ".avro");
        
    //先將schema信息添加到文件中
    fileWriter.create(schema, file);
    //再開始追加GenericRecord記錄
    for(GenericRecord record : records){
        fileWriter.append(record);
    }
    fileWriter.close();
    return file;
}

3.1 DataFileReader反序列化

從內存中反序列化獲得一條或多條記錄

public static List<GenericRecord> deserializeFromMemory(ByteArrayOutputStream out) throws IOException{
    DatumReader<GenericRecord> datumReader = new SpecificDatumReader<GenericRecord>();
    SeekableByteArrayInput sin = new SeekableByteArrayInput(out.toByteArray());
    DataFileReader<GenericRecord> fileReader = new DataFileReader<GenericRecord>(sin, datumReader);
        
    List<GenericRecord> records = new ArrayList<GenericRecord>();
    while(fileReader.hasNext()){
        records.add(fileReader.next());
    }
    fileReader.close();
    return records;
}

從avro文件中反序列化獲得一條或多條記錄

public static List<GenericRecord> deserializeFromFile(File file) throws IOException{
    DatumReader<GenericRecord> datumReader = new SpecificDatumReader<GenericRecord>();
    DataFileReader<GenericRecord> fileReader = new DataFileReader<GenericRecord>(file, datumReader);
        
    List<GenericRecord> records = new ArrayList<GenericRecord>();
    while(fileReader.hasNext()){
        records.add(fileReader.next());
    }
    fileReader.close();
    return records;
}

3.3 特點

a 序列化后的內容中含有Schema信息
b 反序列化時就不再需要Schema信息，因為序列化的內容中已經含有Schema信息
c 可以以內存為存儲媒介，也可以以文件為存儲媒介