Apache Avro データを解析する方法について話しましょう (例付きで説明)

Apache Avro データを解析するにはどうすればよいですか?この記事では、シリアル化して Avro データを生成する方法、デシリアライズして Avro データを解析する方法、FlinkSQL を使用して Avro データを解析する方法を紹介します。

インターネットの急速な発展に伴い、クラウド コンピューティング、ビッグデータ、人工知能 AI、モノのインターネットなどの最先端テクノロジーがハイテクの主流になりました。電子商取引ウェブサイト、顔認識、無人運転、スマートホーム、スマートシティなどの今日の時代のテクノロジーは、人々の日用品、食料、住居、交通手段を容易にするだけでなく、その舞台裏では常に大きな問題が存在します。さまざまなシステム プラットフォームによって収集、消去、分析されるデータの量が多く、低遅延、高スループット、データのセキュリティを確保することが特に重要です。Apache Avro 自体は、バイナリ送信のためにスキーマを通じてシリアル化されています。データの高速伝送を保証する一方で、データのセキュリティも確保します。avro は現在、さまざまな業界でますます広く使用されています。avro データをどのように処理および解析するかが特に重要です。この記事では、avro データを生成する方法を説明します。 avro データをシリアル化して分析に FlinkSQL を使用します。

この記事は avro 解析のデモです。現在、FlinkSQL は単純な avro データ解析にのみ適しています。複雑なネストされた avro データは当面サポートされていません。

シーンの紹介

この記事では主に次の 3 つの主要な内容を紹介します。

• Avro データをシリアル化して生成する方法

• Avro データを逆シリアル化して解析する方法

• FlinkSQL を使用して Avro データを解析する方法

##前提条件

• avro とは何かを理解するには、Apache avro 公式 Web サイトのクイック スタート ガイドを参照してください。

• avro アプリケーション シナリオを理解する

操作手順

1. 新しい avro maven プロジェクトを作成し、pom 依存関係を構成します

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 ファイルは、クラスへのパス、つまり

$\#\#p\#\#\; が自動的に生成されるように構成されています。\#roject.b\mathrm{}asedir/s\mathrm{\#r}c/m\mathrm{a}\#in/av\mathrm{}ro/and\mathrm{\{project.basedir\}/src/main/avro/and}$ ＃＃プロジェクト＃＃＃＃＃＃。＃

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

user_generic.avro内容如下：

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数据已经生成。

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

avro数据解析成功。

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

hdfs dfs -mkdir -p /tmp/lztest/

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节点执行

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 中国語 Web サイトの他の関連記事を参照してください。