Apache Avro数据的示例分析

这篇文章主要介绍Apache Avro数据的示例分析，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！

随着互联网高速的发展，云计算、大数据、人工智能AI、物联网等前沿技术已然成为当今时代主流的高新技术，诸如电商网站、人脸识别、无人驾驶、智能家居、智慧城市等等，不仅方面方便了人们的衣食住行，背后更是时时刻刻有大量的数据在经过各种各样的系统平台的采集、清晰、分析，而保证数据的低时延、高吞吐、安全性就显得尤为重要，Apache Avro本身通过Schema的方式序列化后进行二进制传输，一方面保证了数据的高速传输，另一方面保证了数据安全性，avro当前在各个行业的应用越来越广泛，如何对avro数据进行处理解析应用就格外重要，本文将演示如果序列化生成avro数据，并使用FlinkSQL进行解析。

本文是avro解析的demo，当前FlinkSQL仅适用于简单的avro数据解析,复杂嵌套avro数据暂时不支持。

场景介绍

本文主要介绍以下三个重点内容：

• 如何序列化生成Avro数据

• 如何反序列化解析Avro数据

• 如何使用FlinkSQL解析Avro数据

前提条件

• 了解avro是什么，可参考apache avro官网快速入门指南

• 了解avro应用场景

操作步骤

新建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}/class="lazy" data-src/main/avro/</sourceDirectory>                            <outputDirectory>${project.basedir}/class="lazy" data-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\mathrm{.}basedir\mathrm{/}src\mathrm{/}main\mathrm{/}avro\mathrm{/}\text{和}$project.basedir/class="lazy" data-src/main/avro/和{project.basedir}/class="lazy" data-src/main/java/，这样配置之后，在执行mvn命令的时候，这个插件就会自动将此目录下的avsc schema生成类文件，并放到后者这个目录下。如果没有生成avro目录，手动创建一下即可。

定义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"]} ]}

编译schema

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

序列化

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

User user1 = new User();user1.setName("Alyssa");user1.setFavoriteNumber(256);// Leave favorite col or null// Alternate constructorUser user2 = new User("Ben", 7, "red");// Construct via builderUser user3 = User.newBuilder()        .setName("Charlie")        .setFavoriteColor("blue")        .setFavoriteNumber(null)        .build();// Serialize user1, user2 and user3 to diskDatumWriter<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内容如下：

Obj
avro.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数据已经生成。

反序列化

通过反序列化代码解析avro数据

// Deserialize Users from diskDatumReader<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数据解析成功。

将user_generic.avro上传至hdfs路径

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

配置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/libchmod 500 flink-sql-avro*.jarchown omm:wheel flink-sql-avro*.jar

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

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

编写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  KafkaTableselect  *from  testHdfs;

保存提交任务

查看对应topic中是否有数据

FlinkSQL解析avro数据成功。

以上是“Apache Avro数据的示例分析”这篇文章的所有内容，感谢各位的阅读！希望分享的内容对大家有帮助，更多相关知识，欢迎关注编程网行业资讯频道！