摘要：本文将演示如果序列化生成 avro 数据，并使用 FlinkSQL 进行解析。
本文分享自华为云社区《【技术分享】Apache Avro数据的序列化、反序列&&FlinkSQL解析Avro数据》，作者： 南派三叔。
技术背景
随着互联网高速的发展，云计算、大数据、人工智能 AI、物联网等前沿技术已然成为当今时代主流的高新技术，诸如电商网站、人脸识别、无人驾驶、智能家居、智慧城市等等，不仅方面方便了人们的衣食住行，背后更是时时刻刻有大量的数据在经过各种各样的系统平台的采集、清晰、分析，而保证数据的低时延、高吞吐、安全性就显得尤为重要，Apache Avro 本身通过 Schema 的方式序列化后进行二进制传输，一方面保证了数据的高速传输，另一方面保证了数据安全性，avro 当前在各个行业的应用越来越广泛，如何对 avro 数据进行处理解析应用就格外重要，本文将演示如果序列化生成 avro 数据，并使用 FlinkSQL 进行解析。
本文是 avro 解析的 demo，当前 FlinkSQL 仅适用于简单的 avro 数据解析,复杂嵌套 avro 数据暂时不支持。
场景介绍
本文主要介绍以下三个重点内容：
如何序列化生成 Avro 数据
如何反序列化解析 Avro 数据
如何使用 FlinkSQL 解析 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 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 内容如下：
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 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 数据解析成功。
6、将 user_generic.avro 上传至 hdfs 路径
hdfs dfs -mkdir -p /tmp/lztest/
hdfs dfs -put user_generic.avro /tmp/lztest/
复制代码
7、配置 flinkserver
将 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
复制代码
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 KafkaTableselect *from testHdfs;
复制代码
保存提交任务
9、查看对应 topic 中是否有数据
FlinkSQL 解析 avro 数据成功。
点击关注，第一时间了解华为云新鲜技术~
评论