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

Artikel
09/20/2024

Wichtig

Azure HDInsight auf AKS wurde am 31. Januar 2025 eingestellt. Erfahren Sie mehr mit dieser Ankündigung.

Sie müssen Ihre Workloads zu Microsoft Fabric oder ein gleichwertiges Azure-Produkt migrieren, um eine abrupte Beendigung Ihrer Workloads zu vermeiden.

Wichtig

Dieses Feature befindet sich derzeit in der Vorschau. Die zusätzlichen Nutzungsbedingungen für Microsoft Azure Previews weitere rechtliche Bestimmungen enthalten, die für Azure-Features gelten, die in der Betaversion, in der Vorschau oder auf andere Weise noch nicht in die allgemeine Verfügbarkeit veröffentlicht werden. Informationen zu dieser spezifischen Vorschau finden Sie unter Azure HDInsight auf AKS-Vorschauinformationen. Für Fragen oder Funktionsvorschläge senden Sie bitte eine Anfrage mit den Details an AskHDInsight und folgen Sie uns, um weitere Updates zur Azure HDInsight Communityzu erhalten.

In diesem Artikel erfahren Sie, wie Sie die Echtzeitereignisse bereichern können, indem Sie einen Stream von Kafka mit der Tabelle "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 Attribute-verknüpfte Ereignisse, um in ein anderes Kafka-Thema einzufangen.

Voraussetzungen

Flink-Cluster auf HDInsight auf AKS
Kafka-Cluster auf HDInsight
- Stellen Sie sicher, dass die Netzwerkeinstellungen, wie unter Verwenden von Kafka auf HDInsight beschrieben, vorgenommen werden, indem sichergestellt wird, dass sich HDInsight auf AKS und HDInsight-Clustern im selben 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

Datei für ADLS Gen2 vorbereiten

Wir erstellen eine Datei namens item attributes in unserem Speicher

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

Entwickeln des Apache Flink-Auftrags

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

Erweitern Sie das Thema user_events von Kafka, indem Sie es mit item attributes aus einer Datei auf ADLS Gen2 verbinden.
Wir übertragen das Ergebnis dieses Schritts in Form einer angereicherten Benutzeraktivität mit 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>

Dem Kafka-Thema mit ADLS Gen2 File beitreten

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-Paket erstellen und bei Apache Flink einreichen

Wir übermitteln die verpackte Jar-Datei an Flink.

Echtzeit-`user_events`-Topic auf Kafka produzieren

Wir können echtzeitbasiertes Benutzerverhaltensereignis user_events in Kafka erzeugen.

Verbinden Sie `itemAttributes` mit `user_events` auf Kafka

Wir verwenden jetzt itemAttributes bei Dateisystembeitrittsereignissen im Rahmen von Benutzeraktivitäten user_events.

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

Referenz

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

Freigeben über

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

Voraussetzungen

Kafka Themenvorbereitung

Datei für ADLS Gen2 vorbereiten

Entwickeln des Apache Flink-Auftrags

Entwickeln eines Maven-Projekts

Jar-Paket erstellen und bei Apache Flink einreichen

Echtzeit-`user_events`-Topic auf Kafka produzieren

Verbinden Sie `itemAttributes` mit `user_events` auf Kafka

Referenz

Zusätzliche Ressourcen

Freigeben über

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

Voraussetzungen

Kafka Themenvorbereitung

Datei für ADLS Gen2 vorbereiten

Entwickeln des Apache Flink-Auftrags

Entwickeln eines Maven-Projekts

Jar-Paket erstellen und bei Apache Flink einreichen

Echtzeit-user_events-Topic auf Kafka produzieren

Verbinden Sie itemAttributes mit user_events auf Kafka

Referenz

Zusätzliche Ressourcen

Echtzeit-`user_events`-Topic auf Kafka produzieren

Verbinden Sie `itemAttributes` mit `user_events` auf Kafka