一文解析Apache Avro数据 _数据

学向勤中得，萤窗万卷书。这篇文章主要讲述一文解析Apache Avro数据相关的知识，希望能为你提供帮助。

摘要：本文将演示如果序列化生成avro数据，并使用FlinkSQL进行解析。

本文分享自华为云社区??《【技术分享】Apache Avro数据的序列化、反序列& & FlinkSQL解析Avro数据》??，作者：南派三叔。
技术背景随着互联网高速的发展，云计算、大数据、人工智能AI、物联网等前沿技术已然成为当今时代主流的高新技术，诸如电商网站、人脸识别、无人驾驶、智能家居、智慧城市等等，不仅方面方便了人们的衣食住行，背后更是时时刻刻有大量的数据在经过各种各样的系统平台的采集、清晰、分析，而保证数据的低时延、高吞吐、安全性就显得尤为重要，Apache Avro本身通过Schema的方式序列化后进行二进制传输，一方面保证了数据的高速传输，另一方面保证了数据安全性，avro当前在各个行业的应用越来越广泛，如何对avro数据进行处理解析应用就格外重要，本文将演示如果序列化生成avro数据，并使用FlinkSQL进行解析。
本文是avro解析的demo，当前FlinkSQL仅适用于简单的avro数据解析,复杂嵌套avro数据暂时不支持。
场景介绍本文主要介绍以下三个重点内容：

如何序列化生成Avro数据
如何反序列化解析Avro数据
如何使用FlinkSQL解析Avro数据

前提条件

了解avro是什么，可参考apache avro官网快速入门指南
了解avro应用场景

操作步骤1、新建avro maven工程项目，配置pom依赖

pom文件内容如下：

< ?xml version="1.0" encoding="UTF-8"?>
< project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
< modelVersion> 4.0.0< /modelVersion>

< groupId> com.huawei.bigdata< /groupId>
< artifactId> avrodemo< /artifactId>
< version> 1.0-SNAPSHOT< /version>
< dependencies>
< dependency>
< groupId> org.apache.avro< /groupId>
< artifactId> avro< /artifactId>
< version> 1.8.1< /version>
< /dependency>
< dependency>
< groupId> junit< /groupId>
< artifactId> junit< /artifactId>
< version> 4.12< /version>
< /dependency>
< /dependencies>

< build>
< plugins>
< plugin>
< groupId> org.apache.avro< /groupId>
< artifactId> avro-maven-plugin< /artifactId>
< version> 1.8.1< /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>
< /plugins>
< /build>

< /project>

注意：以上pom文件配置了自动生成类的路径，即$project.basedir/src/main/avro/和$project.basedir/src/main/java/，这样配置之后，在执行mvn命令的时候，这个插件就会自动将此目录下的avsc schema生成类文件，并放到后者这个目录下。如果没有生成avro目录，手动创建一下即可。
2、定义schema
使用JSON为Avro定义schema。schema由基本类型（null,boolean, int, long, float, double, bytes 和string）和复杂类型（record, enum, array, map, union, 和fixed）组成。例如，以下定义一个user的schema，在main目录下创建一个avro目录，然后在avro目录下新建文件 user.avsc :

"namespace": "lancoo.ecbdc.pre",
"type": "record",
"name": "User",
"fields": [
"name": "name", "type": "string",
"name": "favorite_number","type": ["int", "null"],
"name": "favorite_color", "type": ["string", "null"]
]

3、编译schema
点击maven projects项目的compile进行编译，会自动在创建namespace路径和User类代码

4、序列化
创建TestUser类，用于序列化生成数据

User user1 = new User();
user1.setName("Alyssa");
user1.setFavoriteNumber(256);
// Leave favorite col or null

// Alternate constructor
User user2 = new User("Ben", 7, "red");

// Construct via builder
User user3 = User.newBuilder()
.setName("Charlie")
.setFavoriteColor("blue")
.setFavoriteNumber(null)
.build();

// Serialize user1, user2 and user3 to disk
DatumWriter< User> userDatumWriter = new SpecificDatumWriter< User> (User.class);
DataFileWriter< User> dataFileWriter = new DataFileWriter< User> (userDatumWriter);
dataFileWriter.create(user1.getSchema(), new File("user_generic.avro"));
dataFileWriter.append(user1);
dataFileWriter.append(user2);
dataFileWriter.append(user3);
dataFileWriter.close();

执行序列化程序后，会在项目的同级目录下生成avro数据

user_generic.avro内容如下：

Objavro.schema?"type":"record","name":"User","namespace":"lancoo.ecbdc.pre","fields":["name":"name","type":"string","name":"favorite_number","type":["int","null"],"name":"favorite_color","type":["string","null"]]

至此avro数据已经生成。
5、反序列化
通过反序列化代码解析avro数据

// Deserialize Users from disk
DatumReader< User> userDatumReader = new SpecificDatumReader< User> (User.class);
DataFileReader< User> dataFileReader = new DataFileReader< User> (new File("user_generic.avro"), userDatumReader);
User user = null;
while (dataFileReader.hasNext())
// Reuse user object by passing it to next(). This saves us from
// allocating and garbage collecting many objects for files with
// many items.
user = dataFileReader.next(user);
System.out.println(user);

执行反序列化代码解析user_generic.avro

avro数据解析成功。
6、将user_generic.avro上传至hdfs路径

hdfs dfs -mkdir -p /tmp/lztest/

hdfs dfs -put user_generic.avro /tmp/lztest/

7、配置flinkserver

准备avro jar包

将flink-sql-avro-*.jar、flink-sql-avro-confluent-registry-*.jar放入flinkserver lib，将下面的命令在所有flinkserver节点执行

cp /opt/huawei/Bigdata/FusionInsight_Flink_8.1.2/install/FusionInsight-Flink-1.12.2/flink/opt/flink-sql-avro*.jar /opt/huawei/Bigdata/FusionInsight_Flink_8.1.3/install/FusionInsight-Flink-1.12.2/flink/lib

chmod 500 flink-sql-avro*.jar

chown omm:wheel flink-sql-avro*.jar

同时重启FlinkServer实例，重启完成后查看avro包是否被上传

hdfs dfs -ls /FusionInsight_FlinkServer/8.1.2-312005/lib

8、编写FlinkSQL

CREATE TABLE testHdfs(
name String,
favorite_number int,
favorite_color String
) WITH(
connector = filesystem,
path = hdfs:///tmp/lztest/user_generic.avro,
format = avro
); CREATE TABLE KafkaTable (
name String,
favorite_number int,
favorite_color String
) WITH (
connector = kafka,
topic = testavro,
properties.bootstrap.servers = 96.10.2.1:21005,
properties.group.id = testGroup,
scan.startup.mode = latest-offset,
format = avro
);
insert into
KafkaTable
select
*
from
testHdfs;