聊聊怎麼解析Apache Avro資料（範例講解）-Apache-PHP中文網

怎麼解析Apache Avro資料？這篇文章跟大家介紹一下序列化產生Avro資料、反序列化解析Avro資料、使用FlinkSQL解析Avro資料的方法，希望對大家有幫助！聊聊怎麼解析Apache 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依賴聊聊怎麼解析Apache Avro資料（範例講解）

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檔案設定了自動生成類別的路徑，即

a

和

聊聊怎麼解析Apache 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类代码

聊聊怎麼解析Apache Avro資料（範例講解）

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

聊聊怎麼解析Apache 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
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

聊聊怎麼解析Apache Avro資料（範例講解）

avro数据解析成功。

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

hdfs dfs -mkdir -p /tmp/lztest/

hdfs dfs -put user_generic.avro /tmp/lztest/

登入後複製

聊聊怎麼解析Apache Avro資料（範例講解）

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

登入後複製

聊聊怎麼解析Apache Avro資料（範例講解）

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

聊聊怎麼解析Apache Avro資料（範例講解）

8、编写FlinkSQL

CREATE TABLE testHdfs(
  name String,
  favorite_number int,
  favorite_color String
) WITH(
  &#39;connector&#39; = &#39;filesystem&#39;,
  &#39;path&#39; = &#39;hdfs:///tmp/lztest/user_generic.avro&#39;,
  &#39;format&#39; = &#39;avro&#39;
);CREATE TABLE KafkaTable (
  name String,
  favorite_number int,
  favorite_color String
) WITH (
  &#39;connector&#39; = &#39;kafka&#39;,
  &#39;topic&#39; = &#39;testavro&#39;,
  &#39;properties.bootstrap.servers&#39; = &#39;96.10.2.1:21005&#39;,
  &#39;properties.group.id&#39; = &#39;testGroup&#39;,
  &#39;scan.startup.mode&#39; = &#39;latest-offset&#39;,
  &#39;format&#39; = &#39;avro&#39;
);
insert into
  KafkaTable
select
  *
from
  testHdfs;

登入後複製

聊聊怎麼解析Apache Avro資料（範例講解）