Reichern Sie die Ereignisse von Apache Kafka® mit Attributen von ADLS Gen2 mit Apache Flink® an

Artikel
10/24/2024

Hinweis

Azure HDInsight on AKS wird am 31. Januar 2025 eingestellt. Vor dem 31. Januar 2025 müssen Sie Ihre Workloads zu Microsoft Fabric oder einem gleichwertigen Azure-Produkt migrieren, um eine abruptes Beendigung Ihrer Workloads zu vermeiden. Die verbleibenden Cluster in Ihrem Abonnement werden beendet und vom Host entfernt.

Bis zum Einstellungsdatum ist nur grundlegende Unterstützung verfügbar.

Wichtig

Diese Funktion steht derzeit als Vorschau zur Verfügung. Die zusätzlichen Nutzungsbedingungen für Microsoft Azure-Vorschauen enthalten weitere rechtliche Bestimmungen, die für Azure-Features in Betaversionen, in Vorschauversionen oder anderen Versionen gelten, die noch nicht allgemein verfügbar gemacht wurden. Informationen zu dieser spezifischen Vorschau finden Sie unter Informationen zur Vorschau von Azure HDInsight on AKS. Bei Fragen oder Funktionsvorschlägen senden Sie eine Anfrage an AskHDInsight mit den entsprechenden Details, und folgen Sie uns für weitere Updates in der Azure HDInsight-Community.

In diesem Artikel erfahren Sie, wie Sie die Echtzeitereignisse anreichern können, indem Sie einen Stream von Kafka mit der Tabelle auf ADLS Gen2 mit Flink Streaming verknüpfen. Wir verwenden die Flink Streaming-API, um Ereignisse von HDInsight Kafka mit Attributen von ADLS Gen2 zu verknüpfen. Darüber hinaus verwenden wir mit Attributen verknüpfte Ereignisse, um in ein anderes Kafka-Thema einzufangen.

Voraussetzungen

Flink-Cluster in HDInsight on AKS
Kafka-Cluster in HDInsight
- Stellen Sie sicher, dass die Netzwerkeinstellungen wie unter Verwendung von Kafka in HDInsight beschrieben vorgenommen wurden. Dadurch wird sichergestellt, dass sich HDInsight on AKS- und HDInsight-Cluster im gleichen VNet befinden
Für diese Demonstration verwenden wir eine Windows-VM als Maven-Projektentwicklungsumgebung im selben VNet wie HDInsight auf AKS

Kafka-Themenvorbereitung

Wir erstellen ein Thema namens user_events.

Der Zweck besteht darin, einen Datenstrom von Echtzeitereignissen aus einem Kafka-Thema mithilfe von Flink zu lesen. Wir haben jedes Ereignis mit den folgenden Feldern:
```
user_id,
item_id, 
type, 
timestamp, 
```

Kafka 3.2.0

/usr/hdp/current/kafka-broker/bin/kafka-topics.sh --create --replication-factor 2 --partitions 3 --topic user_events --bootstrap-server wn0-contsk:9092
/usr/hdp/current/kafka-broker/bin/kafka-topics.sh --create --replication-factor 2 --partitions 3 --topic user_events_output --bootstrap-server wn0-contsk:9092

Vorbereiten der Datei auf ADLS Gen2

Wir erstellen eine Datei namens item attributes in unserem Speicher

Der Zweck besteht darin, ein Batch von item attributes aus einer Datei auf ADLS Gen2 zu lesen. Jedes Element weist die folgenden Felder:
```
item_id, 
brand, 
category, 
timestamp, 
```

Entwickeln des Apache Flink-Auftrags

In diesem Schritt führen wir die folgenden Aktivitäten aus

Reichern Sie das user_events-Thema von Kafka an, indem Sie mit item attributes aus einer Datei auf ADLS Gen2 verknüpfen.
Wir pushen das Ergebnis dieses Schritts als angereicherte Benutzeraktivität von Ereignissen in ein Kafka-Thema.

Entwickeln eines Maven-Projekts

pom.xml

<?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>contoso.example</groupId>
    <artifactId>FlinkKafkaJoinGen2</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <flink.version>1.17.0</flink.version>
        <java.version>1.8</java.version>
        <scala.binary.version>2.12</scala.binary.version>
        <kafka.version>3.2.0</kafka.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-files -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-files</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka</artifactId>
            <version>${flink.version}</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.0.0</version>
                <configuration>
                    <appendAssemblyId>false</appendAssemblyId>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

Verknüpfen des Kafka-Themas mit ADLS Gen2-Datei

KafkaJoinGen2Demo.java

package contoso.example;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.HashMap;
import java.util.Map;

public class KafkaJoinGen2Demo {
    public static void main(String[] args) throws Exception {
        // 1. Set up the stream execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Kafka source configuration, update with your broker IPs
        String brokers = "<broker-ip>:9092,<broker-ip>:9092,<broker-ip>:9092";
        String inputTopic = "user_events";
        String outputTopic = "user_events_output";
        String groupId = "my_group";

        // 2. Register the cached file, update your container name and storage name
        env.registerCachedFile("abfs://<container-name>@<storagename>.dfs.core.windows.net/flink/data/item.txt", "file1");

        // 3. Read a stream of real-time user behavior event from a Kafka topic
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers(brokers)
                .setTopics(inputTopic)
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStream<String> kafkaData = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");

        // Parse Kafka source data
      DataStream<Tuple4<String, String, String, String>> userEvents = kafkaData.map(new MapFunction<String, Tuple4<String, String, String, String>>() {
          @Override
          public Tuple4<String, String, String, String> map(String value) throws Exception {
              // Parse the line into a Tuple4
              String[] parts = value.split(",");
              if (parts.length < 4) {
                  // Log and skip malformed record
                  System.out.println("Malformed record: " + value);
                  return null;
              }
              return new Tuple4<>(parts[0], parts[1], parts[2], parts[3]);
           }
       });

        // 4. Enrich the user activity events by joining the items' attributes from a file
        DataStream<Tuple7<String,String,String,String,String,String,String>> enrichedData = userEvents.map(new MyJoinFunction());

        // 5. Output the enriched user activity events to a Kafka topic
        KafkaSink<String> sink = KafkaSink.<String>builder()
                .setBootstrapServers(brokers)
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic(outputTopic)
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build()
                )
                .build();

        enrichedData.map(value -> value.toString()).sinkTo(sink);

        // 6. Execute the Flink job
        env.execute("Kafka Join Batch gen2 file, sink to another Kafka Topic");
    }

    private static class MyJoinFunction extends RichMapFunction<Tuple4<String,String,String,String>, Tuple7<String,String,String,String,String,String,String>> {
        private Map<String, Tuple4<String, String, String, String>> itemAttributes;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);

            // Read the cached file and parse its contents into a map
            itemAttributes = new HashMap<>();
            try (BufferedReader reader = new BufferedReader(new FileReader(getRuntimeContext().getDistributedCache().getFile("file1")))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    String[] parts = line.split(",");
                    itemAttributes.put(parts[0], new Tuple4<>(parts[0], parts[1], parts[2], parts[3]));
                }
            }
        }

        @Override
        public Tuple7<String,String,String,String,String,String,String> map(Tuple4<String,String,String,String> value) throws Exception {
            Tuple4<String, String, String, String> broadcastValue = itemAttributes.get(value.f1);
            if (broadcastValue != null) {
                return Tuple7.of(value.f0,value.f1,value.f2,value.f3,broadcastValue.f1,broadcastValue.f2,broadcastValue.f3);
            } else {
                return null;
            }
        }
    }
}

jar verpacken und an Apache Flink übermitteln

Wir übermitteln die verpackte jar an Flink:

Erstellen eines Echtzeit-`user_events`-Themas auf Kafka

Wir sind in der Lage, das Benutzerverhaltensereignis user_events in Echtzeit in Kafka zu erzeugen.

Nutzen der `itemAttributes`-Verknüpfung mit `user_events` auf Kafka

Wir verwenden jetzt itemAttributes bei Dateisystem-Verknüpfungs-Benutzeraktivitätsereignissen user_events.

Wir produzieren und nutzen weiterhin die Benutzeraktivität und Elementattribute in den folgenden Bildern

Verweis

Flink-Beispiele
Apache Flink Website
Apache, Apache Kafka, Kafka, Apache Flink, Flink und zugehörige Open Source-Projektnamen sind Handelsmarken der Apache Software Foundation (ASF).

Freigeben über

Reichern Sie die Ereignisse von Apache Kafka® mit Attributen von ADLS Gen2 mit Apache Flink® an

Voraussetzungen

Kafka-Themenvorbereitung

Vorbereiten der Datei auf ADLS Gen2

Entwickeln des Apache Flink-Auftrags

Entwickeln eines Maven-Projekts

jar verpacken und an Apache Flink übermitteln

Erstellen eines Echtzeit-`user_events`-Themas auf Kafka

Nutzen der `itemAttributes`-Verknüpfung mit `user_events` auf Kafka

Verweis

Feedback

Zusätzliche Ressourcen

Freigeben über

Reichern Sie die Ereignisse von Apache Kafka® mit Attributen von ADLS Gen2 mit Apache Flink® an

Voraussetzungen

Kafka-Themenvorbereitung

Vorbereiten der Datei auf ADLS Gen2

Entwickeln des Apache Flink-Auftrags

Entwickeln eines Maven-Projekts

jar verpacken und an Apache Flink übermitteln

Erstellen eines Echtzeit-user_events-Themas auf Kafka

Nutzen der itemAttributes-Verknüpfung mit user_events auf Kafka

Verweis

Feedback

Zusätzliche Ressourcen

Erstellen eines Echtzeit-`user_events`-Themas auf Kafka

Nutzen der `itemAttributes`-Verknüpfung mit `user_events` auf Kafka