脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → Apache Avro数据

一文解析Apache Avro数据

 更新时间：2022年01月08日 11:02:30   作者：华为云开发者社区  
本文是avro解析的demo，当前FlinkSQL仅适用于简单的avro数据解析,复杂嵌套avro数据暂时不支持。本文主要解析Apache Avro数据的相关内容，感兴趣的朋友一起看看吧

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

 Avro官方文档所写，http://avro.apache.org/docs/current/index.html.

Avro简介

avro是一个数据序列化系统

提供了：

  • 丰富的数据结构
  • 紧凑的，快速的，二进制的数据格式
  • 一种文件格式，用于存储持久化数据
  • 远程过程调用系统（RPC）
  • 和动态语言的简单交互。并不需要为数据文件读写产生代码，也不需要使用或实现RPC协议。代码生成是一种优化方式，但是只对于静态语言有意义。

技术背景

随着互联网高速的发展，云计算、大数据、人工智能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"]}]}

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;

保存提交任务

9、查看对应topic中是否有数据

FlinkSQL解析avro数据成功。

到此这篇关于一文解析Apache Avro数据的文章就介绍到这了,更多相关Apache Avro数据内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • Java并发编程之ReadWriteLock读写锁的操作方法

    Java并发编程之ReadWriteLock读写锁的操作方法

    这篇文章主要介绍了Java并发编程之ReadWriteLock读写锁的操作方法,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
    2021-02-02
  • Java中final与继承操作实例分析

    Java中final与继承操作实例分析

    这篇文章主要介绍了Java中final与继承操作,结合实例形式分析了Java中使用final阻止继承的相关原理与操作注意事项,需要的朋友可以参考下
    2019-09-09
  • Java内部类及其特点的讲解

    Java内部类及其特点的讲解

    今天小编就为大家分享一篇关于Java内部类及其特点的讲解，小编觉得内容挺不错的，现在分享给大家，具有很好的参考价值，需要的朋友一起跟随小编来看看吧
    2019-01-01
  • Java中final关键字详解及实例

    Java中final关键字详解及实例

    这篇文章主要介绍了Java中final关键字详解及实例，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2019-03-03
  • Java使用agent实现main方法之前的实例详解

    Java使用agent实现main方法之前的实例详解

    这篇文章主要介绍了Java使用agent实现main方法之前的实例详解的相关资料,希望通过本文能帮助到大家，让大家理解这部分内容，需要的朋友可以参考下
    2017-10-10
  • SpringBoot基于SpringSecurity表单登录和权限验证的示例

    SpringBoot基于SpringSecurity表单登录和权限验证的示例

    这篇文章主要介绍了SpringBoot基于SpringSecurity表单登录和权限验证的示例。文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2020-09-09
  • 解决SpringBoot集成Eureka导致返回结果由json变为xml的问题

    解决SpringBoot集成Eureka导致返回结果由json变为xml的问题

    这篇文章主要介绍了解决SpringBoot集成Eureka导致返回结果由json变为xml的问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2021-07-07
  • JAVA文件读写例题实现过程解析

    JAVA文件读写例题实现过程解析

    这篇文章主要介绍了JAVA文件读写例题实现过程解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-06-06
  • Spring中基于xml的AOP的详细步骤

    Spring中基于xml的AOP的详细步骤

    这篇文章主要介绍了Spring中基于xml的AOP的详细步骤，本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2020-06-06
  • SpringBoot中支持Https协议的实现

    SpringBoot中支持Https协议的实现

    本文主要介绍了SpringBoot中支持Https协议的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2023-01-01

最新评论