一文解析Apache Avro数据

转x33g5p2x  于2022-01-04 转载在 其他  
字(6.0k)|赞(0)|评价(0)|浏览(455)

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

本文分享自华为云社区《【技术分享】Apache Avro数据的序列化、反序列&&FlinkSQL解析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文件内容如下：

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  7.     <groupId>com.huawei.bigdata</groupId>
  8.     <artifactId>avrodemo</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  10.     <dependencies>
  11.         <dependency>
  12.             <groupId>org.apache.avro</groupId>
  13.             <artifactId>avro</artifactId>
  14.             <version>1.8.1</version>
  15.         </dependency>
  16.         <dependency>
  17.             <groupId>junit</groupId>
  18.             <artifactId>junit</artifactId>
  19.             <version>4.12</version>
  20.         </dependency>
  21.     </dependencies>
  23.     <build>
  24.         <plugins>
  25.             <plugin>
  26.                 <groupId>org.apache.avro</groupId>
  27.                 <artifactId>avro-maven-plugin</artifactId>
  28.                 <version>1.8.1</version>
  29.                 <executions>
  30.                     <execution>
  31.                         <phase>generate-sources</phase>
  32.                         <goals>
  33.                             <goal>schema</goal>
  34.                         </goals>
  35.                         <configuration>
  36.                             <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
  37.                             <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
  38.                         </configuration>
  39.                     </execution>
  40.                 </executions>
  41.             </plugin>
  42.             <plugin>
  43.                 <groupId>org.apache.maven.plugins</groupId>
  44.                 <artifactId>maven-compiler-plugin</artifactId>
  45.                 <configuration>
  46.                     <source>1.6</source>
  47.                     <target>1.6</target>
  48.                 </configuration>
  49.             </plugin>
  50.         </plugins>
  51.     </build>
  53. </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 :

  1. {"namespace": "lancoo.ecbdc.pre",
  2.  "type": "record",
  3.  "name": "User",
  4.  "fields": [
  5.      {"name": "name", "type": "string"},
  6.      {"name": "favorite_number",  "type": ["int", "null"]},
  7.      {"name": "favorite_color", "type": ["string", "null"]}
  8.  ]
  9. }

3、编译schema

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

4、序列化

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

  1. User user1 = new User();
  2. user1.setName("Alyssa");
  3. user1.setFavoriteNumber(256);
  4. // Leave favorite col or null
  6. // Alternate constructor
  7. User user2 = new User("Ben", 7, "red");
  9. // Construct via builder
  10. User user3 = User.newBuilder()
  11.         .setName("Charlie")
  12.         .setFavoriteColor("blue")
  13.         .setFavoriteNumber(null)
  14.         .build();
  16. // Serialize user1, user2 and user3 to disk
  17. DatumWriter<User> userDatumWriter = new SpecificDatumWriter<User>(User.class);
  18. DataFileWriter<User> dataFileWriter = new DataFileWriter<User>(userDatumWriter);
  19. dataFileWriter.create(user1.getSchema(), new File("user_generic.avro"));
  20. dataFileWriter.append(user1);
  21. dataFileWriter.append(user2);
  22. dataFileWriter.append(user3);
  23. dataFileWriter.close();

执行序列化程序后，会在项目的同级目录下生成avro数据

user_generic.avro内容如下：

  1. 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数据

  1. // Deserialize Users from disk
  2. DatumReader<User> userDatumReader = new SpecificDatumReader<User>(User.class);
  3. DataFileReader<User> dataFileReader = new DataFileReader<User>(new File("user_generic.avro"), userDatumReader);
  4. User user = null;
  5. while (dataFileReader.hasNext()) {
  6.     // Reuse user object by passing it to next(). This saves us from
  7.     // allocating and garbage collecting many objects for files with
  8.     // many items.
  9.     user = dataFileReader.next(user);
  10.     System.out.println(user);
  11. }

执行反序列化代码解析user_generic.avro

avro数据解析成功。

6、将user_generic.avro上传至hdfs路径

  1. hdfs dfs -mkdir -p /tmp/lztest/
  2. 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节点执行

  1. 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
  2. chmod 500 flink-sql-avro*.jar
  3. chown omm:wheel flink-sql-avro*.jar

  • 同时重启FlinkServer实例，重启完成后查看avro包是否被上传
  1. hdfs dfs -ls /FusionInsight_FlinkServer/8.1.2-312005/lib

8、编写FlinkSQL

  1. CREATE TABLE testHdfs(
  2.   name String,
  3.   favorite_number int,
  4.   favorite_color String
  5. ) WITH(
  6.   'connector' = 'filesystem',
  7.   'path' = 'hdfs:///tmp/lztest/user_generic.avro',
  8.   'format' = 'avro'
  9. );CREATE TABLE KafkaTable (
  10.   name String,
  11.   favorite_number int,
  12.   favorite_color String
  13. ) WITH (
  14.   'connector' = 'kafka',
  15.   'topic' = 'testavro',
  16.   'properties.bootstrap.servers' = '96.10.2.1:21005',
  17.   'properties.group.id' = 'testGroup',
  18.   'scan.startup.mode' = 'latest-offset',
  19.   'format' = 'avro'
  20. );
  21. insert into
  22.   KafkaTable
  23. select
  24.   *
  25. from
  26.   testHdfs;

保存提交任务

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

FlinkSQL解析avro数据成功。

点击关注，第一时间了解华为云新鲜技术~

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://huaweicloud.blog.csdn.net/article/details/122250453

内容来源于网络，如有侵权，请联系作者删除！

avro序列化apache技术交流反序列flink sql

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com