一文解析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数据成功。

商匡云商
Logo
注册新帐户
对比商品
  • 合计 (0)
对比
0
购物车