一文解析Apache?Avro數(shù)據(jù)
摘要：本文將演示如果序列化生成avro數(shù)據(jù)，并使用FlinkSQL進(jìn)行解析。
Avro官方文檔所寫，http://avro.apache.org/docs/current/index.html.
Avro簡介
avro是一個數(shù)據(jù)序列化系統(tǒng)
提供了：
- 豐富的數(shù)據(jù)結(jié)構(gòu)
- 緊湊的，快速的，二進(jìn)制的數(shù)據(jù)格式
- 一種文件格式，用于存儲持久化數(shù)據(jù)
- 遠(yuǎn)程過程調(diào)用系統(tǒng)（RPC）
- 和動態(tài)語言的簡單交互。并不需要為數(shù)據(jù)文件讀寫產(chǎn)生代碼，也不需要使用或?qū)崿F(xiàn)RPC協(xié)議。代碼生成是一種優(yōu)化方式，但是只對于靜態(tài)語言有意義。
技術(shù)背景
隨著互聯(lián)網(wǎng)高速的發(fā)展，云計算、大數(shù)據(jù)、人工智能AI、物聯(lián)網(wǎng)等前沿技術(shù)已然成為當(dāng)今時代主流的高新技術(shù)，諸如電商網(wǎng)站、人臉識別、無人駕駛、智能家居、智慧城市等等，不僅方面方便了人們的衣食住行，背后更是時時刻刻有大量的數(shù)據(jù)在經(jīng)過各種各樣的系統(tǒng)平臺的采集、清晰、分析，而保證數(shù)據(jù)的低時延、高吞吐、安全性就顯得尤為重要，Apache Avro本身通過Schema的方式序列化后進(jìn)行二進(jìn)制傳輸，一方面保證了數(shù)據(jù)的高速傳輸，另一方面保證了數(shù)據(jù)安全性，avro當(dāng)前在各個行業(yè)的應(yīng)用越來越廣泛，如何對avro數(shù)據(jù)進(jìn)行處理解析應(yīng)用就格外重要，本文將演示如果序列化生成avro數(shù)據(jù)，并使用FlinkSQL進(jìn)行解析。
本文是avro解析的demo，當(dāng)前FlinkSQL僅適用于簡單的avro數(shù)據(jù)解析,復(fù)雜嵌套avro數(shù)據(jù)暫時不支持。
場景介紹
本文主要介紹以下三個重點內(nèi)容：
- 如何序列化生成Avro數(shù)據(jù)
- 如何反序列化解析Avro數(shù)據(jù)
- 如何使用FlinkSQL解析Avro數(shù)據(jù)
前提條件
- 了解avro是什么，可參考apache avro官網(wǎng)快速入門指南
- 了解avro應(yīng)用場景
操作步驟
1、新建avro maven工程項目，配置pom依賴
pom文件內(nèi)容如下：
<?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/，這樣配置之后，在執(zhí)行mvn命令的時候，這個插件就會自動將此目錄下的avsc schema生成類文件，并放到后者這個目錄下。如果沒有生成avro目錄，手動創(chuàng)建一下即可。
2、定義schema
使用JSON為Avro定義schema。schema由基本類型（null,boolean, int, long, float, double, bytes 和string）和復(fù)雜類型（record, enum, array, map, union, 和fixed）組成。例如，以下定義一個user的schema，在main目錄下創(chuàng)建一個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進(jìn)行編譯，會自動在創(chuàng)建namespace路徑和User類代碼
4、序列化
創(chuàng)建TestUser類，用于序列化生成數(shù)據(jù)
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();執(zhí)行序列化程序后，會在項目的同級目錄下生成avro數(shù)據(jù)
user_generic.avro內(nèi)容如下：
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數(shù)據(jù)
// 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);
}執(zhí)行反序列化代碼解析user_generic.avro
avro數(shù)據(jù)解析成功。
6、將user_generic.avro上傳至hdfs路徑
hdfs dfs -mkdir -p /tmp/lztest/ hdfs dfs -put user_generic.avro /tmp/lztest/
7、配置flinkserver
準(zhǔn)備avro jar包
將flink-sql-avro-*.jar、flink-sql-avro-confluent-registry-*.jar放入flinkserver lib，將下面的命令在所有flinkserver節(jié)點執(zhí)行
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;
保存提交任務(wù)
9、查看對應(yīng)topic中是否有數(shù)據(jù)
FlinkSQL解析avro數(shù)據(jù)成功。
到此這篇關(guān)于一文解析Apache Avro數(shù)據(jù)的文章就介紹到這了,更多相關(guān)Apache Avro數(shù)據(jù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
SpringBoot整合SpringSecurity認(rèn)證與授權(quán)
在項目開發(fā)中,權(quán)限認(rèn)證是很重要的,尤其是一些管理類的系統(tǒng),對于權(quán)限要求更為嚴(yán)格,本文主要介紹了SpringBoot整合SpringSecurity認(rèn)證與授權(quán),感興趣的可以了解一下2023-11-11
這篇文章主要介紹了logback之如何按日期和大小切分日志問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-06-06
關(guān)于mybatis遇到Integer類型的參數(shù)時動態(tài)sql需要注意條件
這篇文章主要介紹了關(guān)于mybatis遇到Integer類型的參數(shù)時動態(tài)sql需要注意條件，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-03-03
這篇文章主要介紹了改善Java代碼之慎用java動態(tài)編譯,需要的朋友可以參考下2021-04-04
這篇文章主要介紹了深入理解Java之HashMap源碼剖析，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-02-02
SpringBoot?+?layui?框架實現(xiàn)一周免登陸功能示例詳解
這篇文章主要介紹了SpringBoot+layui框架實現(xiàn)一周免登陸功能,本文通過示例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2023-08-08
Java實現(xiàn)二叉樹的建立、計算高度與遞歸輸出操作示例
這篇文章主要介紹了Java實現(xiàn)二叉樹的建立、計算高度與遞歸輸出操作,結(jié)合實例形式分析了Java二叉樹的創(chuàng)建、遍歷、計算等相關(guān)算法實現(xiàn)技巧,需要的朋友可以參考下2019-03-03
這篇文章主要介紹了詳解Java String字符串獲取每一個字符及常用方法，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-09-09
Vector類實現(xiàn)了一個動態(tài)數(shù)組。和ArrayList很相似，但是兩者是不同的Vector是同步訪問的；Vector包含了許多傳統(tǒng)的方法，這些方法不屬于集合框架2023-01-01
Java線程的創(chuàng)建介紹及實現(xiàn)方式示例
這篇文章主要為大家介紹了Java線程的創(chuàng)建介紹及實現(xiàn)方式示例,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-09-09