怎么解析apache avro数据？本篇文章给大家介绍一下序列化生成avro数据、反序列化解析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> <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/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();</user></user></user></user>
执行序列化程序后，会在项目的同级目录下生成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></user></user></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;
保存提交任务
9、查看对应topic中是否有数据
FlinkSQL解析avro数据成功。
【推荐：Apache使用教程】
