怎麼解析Apache Avro資料?這篇文章跟大家介紹一下序列化產生Avro資料、反序列化解析Avro資料、使用FlinkSQL解析Avro資料的方法,希望對大家有幫助!
隨著網路高速的發展,雲端運算、大數據、人工智慧AI、物聯網等尖端技術已然成為當今時代主流的高新技術,電商網站、臉部辨識、無人駕駛、智慧家居、智慧城市等等,不僅方面方便了人們的衣食住行,背後更是時時刻刻有大量的數據在經過各種各樣的系統平台的採集、清晰、分析,而確保資料的低時延、高吞吐、安全性就顯得尤為重要,Apache Avro本身透過Schema的方式序列化後進行二進位傳輸,一方面保證了資料的高速傳輸,另一方面保證了資料安全性,avro目前在各行業的應用越來越廣泛,如何對avro數據進行處理解析應用就格外重要,本文將示範如果序列化產生avro數據,並使用FlinkSQL進行解析。
本文是avro解析的demo,目前FlinkSQL僅適用於簡單的avro資料解析,複雜巢狀avro資料暫時不支援。
本文主要介紹以下三個重點內容:
#如何序列化產生Avro資料
#如何反序列化解析Avro資料
如何使用FlinkSQL解析Avro資料
了解avro是什麼,可參考apache avro官網快速入門指南
#了解avro應用程式場景
注意:以上pom檔案設定了自動生成類別的路徑,即4.0.0 com.huawei.bigdata avrodemo 1.0-SNAPSHOT org.apache.avro avro 1.8.1 junit junit 4.12 org.apache.avro avro-maven-plugin 1.8.1 generate-sources schema ${project.basedir}/src/main/avro/ ${project.basedir}/src/main/java/ org.apache.maven.plugins maven-compiler-plugin 1.6
rr
o/和{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 DatumWriteruserDatumWriter = new SpecificDatumWriter (User.class); DataFileWriter dataFileWriter = new DataFileWriter (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 disk DatumReaderuserDatumReader = new SpecificDatumReader (User.class); DataFileReader dataFileReader = new DataFileReader (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
同时重启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使用教程】
以上是聊聊怎麼解析Apache Avro資料(範例講解)的詳細內容。更多資訊請關注PHP中文網其他相關文章!