聊聊怎么解析Apache Avro数据（示例讲解）_Linux

怎么解析apache avro数据？本篇文章给大家介绍一下序列化生成avro数据、反序列化解析avro数据、使用flinksql解析avro数据的方法，希望对大家有所帮助！

聊聊怎么解析apache 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/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();</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;

登录后复制