ApacheAvro數(shù)據(jù)的示例分析

這篇文章主要介紹Apache Avro數(shù)據(jù)的示例分析，文中介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們一定要看完！

讓客戶滿意是我們工作的目標(biāo)，不斷超越客戶的期望值來自于我們對這個行業(yè)的熱愛。我們立志把好的技術(shù)通過有效、簡單的方式提供給客戶，將通過不懈努力成為客戶在信息化領(lǐng)域值得信任、有價值的長期合作伙伴，公司提供的服務(wù)項目有：域名注冊、網(wǎng)站空間、營銷軟件、網(wǎng)站建設(shè)、天臺網(wǎng)站維護(hù)、網(wǎng)站推廣。

隨著互聯(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\mathrm{.}basedir\mathrm{/}src\mathrm{/}main\mathrm{/}avro\mathrm{/}\text{和}$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)容如下：

Obj
avro.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數(shù)據(jù)已經(jīng)生成。

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ù)成功。

以上是“Apache Avro數(shù)據(jù)的示例分析”這篇文章的所有內(nèi)容，感謝各位的閱讀！希望分享的內(nèi)容對大家有幫助，更多相關(guān)知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！

名稱欄目：ApacheAvro數(shù)據(jù)的示例分析
本文URL：http://bm7419.com/article22/gejcjc.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供網(wǎng)站建設(shè)、定制網(wǎng)站、網(wǎng)站改版、微信公眾號、建站公司、微信小程序

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)