avro序列化详细操作

Posted 2020-10-02 厚积！！

Intellij 15.0.3
Maven
avro 1.8.0

Avro是一个数据序列化系统。
它提供以下：
1 丰富的数据结构类型
2 快速可压缩的二进制数据形式
3 存储持久数据的文件容器
4 远程过程调用RPC
5 简单的动态语言结合功能，Avro和动态语言结合后，读写数据文件和使用RPC协议都不需要生成代码，而代码生成作为一种可选的优化只值得在静态类型语言中实现。
Avro依赖于模式(Schema)。Avro数据的读写操作是很频繁的，而这些操作都需要使用模式，这样就减少写入每个数据资料的开销，使得序列化快速而又轻巧。这种数据及其模式的自我描述方便于动态脚本语言的使用。

下面介绍如果使用avro进行序列化和反序列化的操作
前置条件：
maven项目

1、在pom.xml中添加avro的依赖包和编译插件

1.1 配置avro依赖

在dependencies中配置avro的GAV

<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>1.8.0</version>
</dependency>

• 1
• 2
• 3
• 4
• 5

• 1
• 2
• 3
• 4
• 5

1.2 在build–>plugins下配置编译插件

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.8.0</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>schema</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
                <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>
<plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-compiler-plugin</artifactId>
    <configuration>
        <source>1.6</source>
        <target>1.6</target>
    </configuration>
</plugin>

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25

2、定义avro的scheme

Avro scheme是通过JSON形式来定义的，一般以.avsc结尾（maven插件会去指定目录下获取.avsc结尾的文件并生成成Java文件）

2.1 在src/main目录下新建一个avro文件夹

2.2 在src/main/avro目录下新建一个文件，并保存为user.avsc。文件内容如下：

{"namespace": "cn.md31.avro.test.bean",
 "type": "record",
 "name": "User",
 "fields": [
     {"name": "name", "type": "string"},
     {"name": "favorite_number",  "type": ["int", "null"]},
     {"name": "favorite_color", "type": ["string", "null"]}
 ]
}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9

其中
namespace在java项目中翻译成包名
name是类名
fields就是配置的属性
注意：必须配置type为record

2.3 生成User的java文件

avro提供了一个avro-tools包来生成java文件，可以通过下面命令：

java -jar /path/to/avro-tools-1.8.0.jar compile schema <schema file> <destination>

• 1

• 1

在1.2步中，pom文件中配置了插件，所以在maven项目中，我们只需要执行mvn complie 命令就可以自动生成java文件了。
注：插件会自动搜索src/main/avro目录下的.avsc结尾的文件，并生成java文件到src/main/java包下
因为我们在pom文件中指定了相应目录：

如果使用Intellij，我们直接执行compile即可：

执行之后会在相应目录生成User.java文件

 

1 使用生成的User类初始化User对象

Avro自动生成的User类有三种方式初始化：

1.1 使用构造方法：

User user1 = new User("user1", 10, "red");

1.2 使用setter方法

User user2 = new User();
user2.setName("user2");
user2.setFavoriteNumber(11);
user2.setFavoriteColor("white");

• 1
• 2

1.3 使用build方法

User user3 = User.newBuilder()
        .setName("user3")
        .setFavoriteNumber(12)
        .setFavoriteColor("black")
        .build();

整合到produceUsers方法中：

public static List<User> produceUsers() {
    List<User> userList = new ArrayList<User>();
    // 三种初始化方式
    User user1 = new User("user1", 10, "red");
    User user2 = new User();
    user2.setName("user2");
    user2.setFavoriteNumber(11);
    user2.setFavoriteColor("white");
    User user3 = User.newBuilder()
            .setName("user3")
            .setFavoriteNumber(12)
            .setFavoriteColor("black")
            .build();
    userList.add(user1);
    userList.add(user2);
    userList.add(user3);
    return userList;
}

2 把User对象序列化到文件中

public static void serializeAvroToFile(List<User> userList, String fileName) throws IOException {
    DatumWriter<User> userDatumWriter = new SpecificDatumWriter<User>(User.class);
    DataFileWriter<User> dataFileWriter = new DataFileWriter<User>(userDatumWriter);
    dataFileWriter.create(userList.get(0).getSchema(), new File(fileName));
    for (User user: userList) {
        dataFileWriter.append(user);
    }
    dataFileWriter.close();
}

在main方法中调用上面两个方法

public static void main(String[] args) throws IOException {
    List<User> userList = produceUsers();
    String fileName = "users.avro";
    serializeAvroToFile(userList, fileName);
//        deserializeAvroFromFile(fileName);
//
//        byte[] usersByteArray = serializeAvroToByteArray(userList);
//        deserialzeAvroFromByteArray(usersByteArray);
}

执行main方法，会发现在根目录生成了一个users.avro文件：

3 从文件中反序列化对象

public static void deserializeAvroFromFile(String fileName) throws IOException {
    File file = new File(fileName);
    DatumReader<User> userDatumReader = new SpecificDatumReader<User>(User.class);
    DataFileReader<User> dataFileReader = new DataFileReader<User>(file, userDatumReader);
    User user = null;
    System.out.println("----------------deserializeAvroFromFile-------------------");
    while (dataFileReader.hasNext()) {
        user = dataFileReader.next(user);
        System.out.println(user);
    }
}

执行之后得到结果：

4、序列化对象成byte 数组

public static byte[] serializeAvroToByteArray(List<User> userList) throws IOException {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DatumWriter<User> userDatumWriter = new SpecificDatumWriter<User>(User.class);
        DataFileWriter<User> dataFileWriter = new DataFileWriter<User>(userDatumWriter);
        dataFileWriter.create(userList.get(0).getSchema(), baos);
        for (User user: userList) {
            dataFileWriter.append(user);
        }
        dataFileWriter.close();
        return baos.toByteArray();
    }

5、从byte数组中反序列化成对象

public static void deserialzeAvroFromByteArray(byte[] usersByteArray) throws IOException {
        SeekableByteArrayInput sbai = new SeekableByteArrayInput(usersByteArray);
        DatumReader<User> userDatumReader = new SpecificDatumReader<User>(User.class);
        DataFileReader<User> dataFileReader = new DataFileReader<User>(sbai, userDatumReader);
        System.out.println("----------------deserialzeAvroFromByteArray-------------------");
        User readUser = null;
        while (dataFileReader.hasNext()) {
            readUser = dataFileReader.next(readUser);
            System.out.println(readUser);
        }
    }

以上是关于avro序列化详细操作的主要内容，如果未能解决你的问题，请参考以下文章

Apache Avro：简单的序列化和反序列化操作的Demo

0016-Avro序列化&反序列化和Spark读取Avro数据

反序列化 Avro Spark

节Avro序列化的使用

在 Python 中使用“SchemaRegistryClient”反序列化 AVRO 消息

反序列化Avro序列化Kafka流的问题