Wie analysiere ich Apache Avro-Daten? In diesem Artikel werden Ihnen die Methoden der Serialisierung zum Generieren von Avro-Daten, des Deserialisierens zum Parsen von Avro-Daten und die Verwendung von FlinkSQL zum Parsen von Avro-Daten vorgestellt.
Mit der rasanten Entwicklung des Internets sind Spitzentechnologien wie Cloud Computing, Big Data, künstliche Intelligenz und das Internet der Dinge in der heutigen Zeit zu Mainstream-Hightech-Technologien geworden, beispielsweise E-Commerce-Websites , Gesichtserkennung, fahrerlose Autos und Smartphones erleichtern nicht nur die Ernährung, Kleidung, Unterkunft und den Transport der Menschen, sondern dahinter stehen auch große Datenmengen, die von gesammelt, gelöscht und analysiert werden Dabei ist es besonders wichtig, dass Apache Avro selbst über Schema für die Binärübertragung serialisiert wird Andererseits wird Avro derzeit in verschiedenen Branchen immer häufiger verwendet. Je umfangreicher es ist, desto wichtiger ist die Verarbeitung und Analyse von Avro-Daten Verwenden Sie FlinkSQL zur Analyse.
Dieser Artikel ist eine Demo des Avro-Parsings. Derzeit ist FlinkSQL nur für das einfache Parsen von Avro-Daten geeignet.
Szeneneinführung
In diesem Artikel werden hauptsächlich die folgenden drei Hauptinhalte vorgestellt:
So serialisieren und generieren Sie Avro-Daten
So deserialisieren und analysieren Sie Avro-Daten
So verwenden Sie FlinkSQL zum Parsen von Avro-Daten
Voraussetzungen
Um zu verstehen, was Avro ist, können Sie sich die Kurzanleitung der offiziellen Apache-Avro-Website ansehen.
Avro-Anwendungsszenarien verstehen POM-Abhängigkeiten projektieren und konfigurieren
Der Inhalt der POM-Datei lautet wie folgt:
<?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>
Hinweis: Die obige POM-Datei ist mit dem Pfad zur automatisch generierten Klasse konfiguriert, d. h.
p3 Kompilieren Sie das 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
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内容如下:
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
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
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使用教程】
Das obige ist der detaillierte Inhalt vonLassen Sie uns darüber sprechen, wie Apache Avro-Daten analysiert werden (Erklärung mit Beispielen).. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!