数据序列化机制-Avro

2024-04-26 02:08
文章标签 数据 机制 序列化 avro

本文主要是介绍数据序列化机制-Avro，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

序列化主要是将内存缓冲区、数据结构或者对象中的数据转换为能够在网路上传输或者持久化存储（比如磁盘）中存储的二进制文件。

1.Avro的特性？

1)与语言无关

2）基于模式：Avro会序列化数据时会将模式写入其中，Avro序列化数据到一个压缩的二进制格式

3）使用类Json的格式来描述数据的结构，并且支持多种语言，像Java, C, C++, C#, Python, and Ruby。

4）序列化速度快且序列化过后数据存储体积小

5）支持多种数据类型

2.Avro的schema

Avro的Schema用JSON表示。Schema定义了简单数据类型和复杂数据类型。

基本类型

其中简单数据类型有以下8种：

类型含义
null没有值
boolean布尔值
int32位有符号整数
long64位有符号整数
float单精度（32位）的IEEE 754浮点数
double双精度（64位）的IEEE 754浮点数
bytes8位无符号字节序列
string字符串

基本类型没有属性，基本类型的名字也就是类型的名字，比如：

{"type": "string"}

复杂类型

Avro提供了6种复杂类型。分别是Record，Enum，Array，Map，Union和Fixed。

Record

Record类型使用的类型名字是 “record”，还支持其它属性的设置：

name：record类型的名字(必填)

namespace：命名空间(可选)

doc：这个类型的文档说明(可选)

aliases：record类型的别名，是个字符串数组(可选)

fields：record类型中的字段，是个对象数组(必填)。每个字段需要以下属性：

  1. name：字段名字(必填)
  2. doc：字段说明文档(可选)
  3. type：一个schema的json对象或者一个类型名字(必填)
  4. default：默认值(可选)
  5. order：排序(可选)，只有3个值ascending(默认)，descending或ignore
  6. aliases：别名，字符串数组(可选)

一个Record类型例子，定义一个元素类型是Long的链表：

{"type": "record", "name": "LongList","aliases": ["LinkedLongs"],                      // old name for this"fields" : [{"name": "value", "type": "long"},             // each element has a long{"name": "next", "type": ["null", "LongList"]} // optional next element]
}

 

Enum

枚举类型的类型名字是”enum”，还支持其它属性的设置：

name：枚举类型的名字(必填)
namespace：命名空间(可选)
aliases：字符串数组，别名(可选)
doc：说明文档(可选)
symbols：字符串数组，所有的枚举值(必填)，不允许重复数据。

一个枚举类型的例子：

{ "type": "enum","name": "Suit","symbols" : ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
}

Array

数组类型的类型名字是”array”并且只支持一个属性：

items：数组元素的schema

一个数组例子：

{"type": "array", "items": "string"}

Map

Map类型的类型名字是”map”并且只支持一个属性：

values：map值的schema

Map的key必须是字符串。

一个Map例子：

{"type": "map", "values": "long"}

Union

组合类型，表示各种类型的组合，使用数组进行组合。比如[“null”, “string”]表示类型可以为null或者string。

组合类型的默认值是看组合类型的第一个元素，因此如果一个组合类型包括null类型，那么null类型一般都会放在第一个位置，这样子的话这个组合类型的默认值就是null。

组合类型中不允许同一种类型的元素的个数不会超过1个，除了record，fixed和enum。比如组合类中有2个array类型或者2个map类型，这是不允许的。

组合类型不允许嵌套组合类型。

Fixed

混合类型的类型名字是fixed，支持以下属性：

name：名字(必填)
namespace：命名空间(可选)
aliases：字符串数组，别名(可选)
size：一个整数，表示每个值的字节数(必填)

比如16个字节数的fixed类型例子如下：

{"type": "fixed", "size": 16, "name": "md5"}

1个Avro例子

首先定义一个User的schema：

{
"namespace": "example.avro","type": "record","name": "User","fields": [{"name": "name", "type": "string"},{"name": "favorite_number",  "type": "int"},{"name": "favorite_color", "type": "string"}]
}

User有3个属性，分别是name，favorite_number和favorite_color。

json文件内容：

{"name":"format","favorite_number":1,"favorite_color":"red"}
{"name":"format2","favorite_number":2,"favorite_color":"black"}
{"name":"format3","favorite_number":666,"favorite_color":"blue"}

使用avro工具将json文件转换成avro文件：

ava -jar avro-tools-1.8.0.jar fromjson --schema-file user.avsc user.json > user.avro

可以设置压缩格式：

java -jar avro-tools-1.8.0.jar fromjson --codec snappy --schema-file user.avsc user.json > user2.avro

将avro文件反转换成json文件：

java -jar avro-tools-1.8.0.jar tojson user.avro
java -jar avro-tools-1.8.0.jar --pretty tojson user.avro

得到avro文件的meta：

java -jar avro-tools-1.8.0.jar getmeta user.avro

输出：

avro.codec    null
avro.schema    {"type":"record","name":"User","namespace":"example.avro","fields":[{"name":"name","type":"string"},{"name":"favorite_number","type":"int"},{"name":"favorite_color","type":"string"}]}

 将文本文件转换成avro文件：

java -jar avro-tools-1.8.0.jar fromtext user.txt usertxt.avro

Avro使用生成的代码进行序列化和反序列化

以上面一个例子的schema为例讲解。

Avro可以根据schema自动生成对应的类：

java -jar /path/to/avro-tools-1.8.0.jar compile schema user.avsc .

user.avsc的namespace为example.avro，name为User。最终在当前目录生成的example/avro目录下有个User.java文件。

├── example │ └── avro │ └── User.java

使用Avro生成的代码创建User：

User user1 = new User();
user1.setName("Format");
user1.setFavoriteColor("red");
user1.setFavoriteNumber(666);User user2 = new User("Format2", 66, "blue");User user3 = User.newBuilder().setName("Format3").setFavoriteNumber(6).setFavoriteColor("black").build();

可以使用有参的构造函数和无参的构造函数，也可以使用Builder构造User。

序列化：

DatumWrite接口用来把java对象转换成内存中的序列化格式，SpecificDatumWriter用来生成类并且指定生成的类型。

最后使用DataFileWriter来进行具体的序列化，create方法指定文件和schema信息，append方法用来写数据，最后写完后close文件

DatumWriter<User> userDatumWriter = new SpecificDatumWriter<User>(User.class);DataFileWriter<User> dataFileWriter = new DataFileWriter<User>(userDatumWriter);
dataFileWriter.create(user1.getSchema(), new File("users.avro"));
dataFileWriter.append(user1);
dataFileWriter.append(user2);
dataFileWriter.append(user3);
dataFileWriter.close();

反序列化：

反序列化跟序列化很像，相应的Writer换成Reader。这里只创建一个User对象是为了性能优化，每次都重用这个User对象，如果文件量很大，对象分配和垃圾收集处理的代价很昂贵。如果不考虑性能，可以使用 for (User user : dataFileReader) 循环遍历对象

File file = new File("users.avro");
DatumReader<User> userDatumReader = new SpecificDatumReader<User>(User.class);
DataFileReader<User> dataFileReader = new DataFileReader<User>(file, userDatumReader);
User user = null;
while(dataFileReader.hasNext()) {user = dataFileReader.next(user);System.out.println(user);
}

打印出：

{"name": "Format", "favorite_number": 666, "favorite_color": "red"}
{"name": "Format2", "favorite_number": 66, "favorite_color": "blue"}
{"name": "Format3", "favorite_number": 6, "favorite_color": "black"}

Avro不使用生成的代码进行序列化和反序列化

虽然Avro为我们提供了根据schema自动生成类的方法，我们也可以自己创建类，不使用Avro的自动生成工具。

创建User：

首先使用Parser读取schema信息并且创建Schema类：

Schema schema = new Schema.Parser().parse(new File("user.avsc"));

有了Schema之后可以创建record：

GenericRecord user1 = new GenericData.Record(schema);
user1.put("name", "Format");
user1.put("favorite_number", 666);
user1.put("favorite_color", "red");GenericRecord user2 = new GenericData.Record(schema);
user2.put("name", "Format2");
user2.put("favorite_number", 66);
user2.put("favorite_color", "blue");

使用GenericRecord表示User，GenericRecord会根据schema验证字段是否正确，如果put进了不存在的字段 user1.put(“favorite_animal”, “cat”) ，那么运行的时候会得到AvroRuntimeException异常。

序列化：

序列化跟生成的User类似，只不过schema是自己构造的，不是User中拿的。

Schema schema = new Schema.Parser().parse(new File("user.avsc"));
GenericRecord user1 = new GenericData.Record(schema);
user1.put("name", "Format");
user1.put("favorite_number", 666);
user1.put("favorite_color", "red");GenericRecord user2 = new GenericData.Record(schema);
user2.put("name", "Format2");
user2.put("favorite_number", 66);
user2.put("favorite_color", "blue");DatumWriter<GenericRecord> datumWriter = new SpecificDatumWriter<GenericRecord>(schema);
DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<GenericRecord>(datumWriter);
dataFileWriter.create(schema, new File("users2.avro"));
dataFileWriter.append(user1);
dataFileWriter.append(user2);
dataFileWriter.close();

反序列化：

反序列化跟生成的User类似，只不过schema是自己构造的，不是User中拿的。

Schema schema = new Schema.Parser().parse(new File("user.avsc"));
File file = new File("users2.avro");
DatumReader<GenericRecord> datumReader = new SpecificDatumReader<GenericRecord>(schema);
DataFileReader<GenericRecord> dataFileReader = new DataFileReader<GenericRecord>(file, datumReader);
GenericRecord user = null;
while(dataFileReader.hasNext()) {user = dataFileReader.next(user);System.out.println(user);
}

打印出：

{"name": "Format", "favorite_number": 666, "favorite_color": "red"}
{"name": "Format2", "favorite_number": 66, "favorite_color": "blue"}

一些注意点

Avro解析json文件的时候，如果类型是Record并且里面有字段是union并且允许空值的话，需要进行转换。因为[“bytes”, “string”]和[“int”,”long”]这2个union类型在json中是有歧义的，第一个union在json中都会被转换成string类型，第二个union在json中都会被转换成数字类型。

所以如果json值的null的话，在avro提供的json中直接写null，否则使用只有一个键值对的对象，键是类型，值的具体的值。

比如：

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

在要转换成json文件的时候要写成这样：

{"name":"format","favorite_number":{"int":1},"favorite_color":{"string":"red"}}
{"name":"format2","favorite_number":null,"favorite_color":{"string":"black"}}
{"name":"format3","favorite_number":{"int":66},"favorite_color":null}

Spark读取Avro文件

直接遍历avro文件，得到GenericRecord进行处理：

val conf = new SparkConf().setMaster("local").setAppName("AvroTest")val sc = new SparkContext(conf)val rdd = sc.hadoopFile[AvroWrapper[GenericRecord], NullWritable, AvroInputFormat[GenericRecord]](this.getClass.getResource("/").toString + "users.avro")val nameRdd = rdd.map(s => s._1.datum().get("name").toString)nameRdd.collect().foreach(println)

使用Avro需要注意的地方

笔者使用Avro的时候暂时遇到了下面2个坑。先记录一下，以后遇到新的坑会更新这篇文章。

1.如果定义了unions类型的字段，而且unions中有null选项的schema，比如如下schema：

{
"namespace": "example.avro","type": "record","name": "User2","fields": [{"name": "name", "type": "string"},{"name": "favorite_number",  "type": ["null","int"]},{"name": "favorite_color", "type": ["null","string"]}]
}

这样的schema，如果不使用Avro自动生成的model代码进行insert，并且insert中的model数据有null数据的话。然后用spark读avro文件的话，会报org.apache.avro.AvroTypeException: Found null, expecting int … 这样的错误。

这一点很奇怪，但是使用Avro生成的Model进行insert的话，sprak读取就没有任何问题。 很困惑。

2.如果使用了Map类型的字段，avro生成的model中的Map的Key默认类型为CharSequence。这种model我们insert数据的话，用String是没有问题的。但是spark读取之后要根据Key拿这个Map数据的时候，永远得到的是null。

stackoverflow上有一个页面说到了这个问题。http://stackoverflow.com/questions/19728853/apache-avro-map-uses-charsequence-as-key

需要在map类型的字段里加上”avro.java.string”: “String”这个选项, 然后compile的时候使用-string参数即可。

比如以下这个schema：

{
"namespace": "example.avro","type": "record","name": "User3","fields": [{"name": "name", "type": "string"},{"name": "favorite_number",  "type": ["null","int"]},{"name": "favorite_color", "type": ["null","string"]},{"name": "scores", "type": ["null", {"type": "map", "values": "string", "avro.java.string": "String"}]}]
}

 

这篇关于数据序列化机制-Avro的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/936431。

相关文章

SpringBoot分段处理List集合多线程批量插入数据方式

SpringBoot分段处理List集合多线程批量插入数据方式

《SpringBoot分段处理List集合多线程批量插入数据方式》文章介绍如何处理大数据量List批量插入数据库的优化方案：通过拆分List并分配独立线程处理,结合Spring线程池与异步方法提升效率... 目录项目场景解决方案1.实体类2.Mapper3.spring容器注入线程池bejsan对象4.创建
阅读更多...
PHP轻松处理千万行数据的方法详解

PHP轻松处理千万行数据的方法详解

《PHP轻松处理千万行数据的方法详解》说到处理大数据集,PHP通常不是第一个想到的语言,但如果你曾经需要处理数百万行数据而不让服务器崩溃或内存耗尽,你就会知道PHP用对了工具有多强大,下面小编就... 目录问题的本质php 中的数据流处理：为什么必不可少生成器：内存高效的迭代方式流量控制：避免系统过载一次性
阅读更多...
C#实现千万数据秒级导入的代码

C#实现千万数据秒级导入的代码

《C#实现千万数据秒级导入的代码》在实际开发中excel导入很常见,现代社会中很容易遇到大数据处理业务,所以本文我就给大家分享一下千万数据秒级导入怎么实现,文中有详细的代码示例供大家参考,需要的朋友可... 目录前言一、数据存储二、处理逻辑优化前代码处理逻辑优化后的代码总结前言在实际开发中excel导入很
阅读更多...
MyBatis-plus处理存储json数据过程

MyBatis-plus处理存储json数据过程

《MyBatis-plus处理存储json数据过程》文章介绍MyBatis-Plus3.4.21处理对象与集合的差异：对象可用内置Handler配合autoResultMap,集合需自定义处理器继承F... 目录1、如果是对象2、如果需要转换的是List集合总结对象和集合分两种情况处理，目前我用的MP的版本
阅读更多...
GSON框架下将百度天气JSON数据转JavaBean

GSON框架下将百度天气JSON数据转JavaBean

《GSON框架下将百度天气JSON数据转JavaBean》这篇文章主要为大家详细介绍了如何在GSON框架下实现将百度天气JSON数据转JavaBean,文中的示例代码讲解详细,感兴趣的小伙伴可以了解下... 目录前言一、百度天气jsON1、请求参数2、返回参数3、属性映射二、GSON属性映射实战1、类对象映
阅读更多...
C# LiteDB处理时间序列数据的高性能解决方案

C# LiteDB处理时间序列数据的高性能解决方案

《C#LiteDB处理时间序列数据的高性能解决方案》LiteDB作为.NET生态下的轻量级嵌入式NoSQL数据库,一直是时间序列处理的优选方案,本文将为大家大家简单介绍一下LiteDB处理时间序列数... 目录为什么选择LiteDB处理时间序列数据第一章：LiteDB时间序列数据模型设计1.1 核心设计原则
阅读更多...
Java+AI驱动实现PDF文件数据提取与解析

Java+AI驱动实现PDF文件数据提取与解析

《Java+AI驱动实现PDF文件数据提取与解析》本文将和大家分享一套基于AI的体检报告智能评估方案,详细介绍从PDF上传、内容提取到AI分析、数据存储的全流程自动化实现方法,感兴趣的可以了解下... 目录一、核心流程：从上传到评估的完整链路二、第一步：解析 PDF，提取体检报告内容1. 引入依赖2. 封装
阅读更多...
基于Redis自动过期的流处理暂停机制

基于Redis自动过期的流处理暂停机制

《基于Redis自动过期的流处理暂停机制》基于Redis自动过期的流处理暂停机制是一种高效、可靠且易于实现的解决方案,防止延时过大的数据影响实时处理自动恢复处理,以避免积压的数据影响实时性,下面就来详... 目录核心思路代码实现1. 初始化Redis连接和键前缀2. 接收数据时检查暂停状态3. 检测到延时过
阅读更多...
Redis中哨兵机制和集群的区别及说明

Redis中哨兵机制和集群的区别及说明

《Redis中哨兵机制和集群的区别及说明》Redis哨兵通过主从复制实现高可用,适用于中小规模数据；集群采用分布式分片,支持动态扩展,适合大规模数据,哨兵管理简单但扩展性弱,集群性能更强但架构复杂,根... 目录一、架构设计与节点角色1. 哨兵机制（Sentinel）2. 集群（Cluster）二、数据分片
阅读更多...
MySQL中查询和展示LONGBLOB类型数据的技巧总结

MySQL中查询和展示LONGBLOB类型数据的技巧总结

《MySQL中查询和展示LONGBLOB类型数据的技巧总结》在MySQL中LONGBLOB是一种二进制大对象（BLOB）数据类型,用于存储大量的二进制数据,：本文主要介绍MySQL中查询和展示LO... 目录前言1. 查询 LONGBLOB 数据的大小2. 查询并展示 LONGBLOB 数据2.1 转换为十
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2