Arricchire gli eventi di Apache Kafka® con attributi di ADLS Gen2 con Apache Flink®

Articolo
09/20/2024

Importante

Azure HDInsight su Azure Kubernetes Service (AKS) è stato ritirato il 31 gennaio 2025. Scopri di più con questo annuncio.

È necessario eseguire la migrazione dei carichi di lavoro a Microsoft Fabric o a un prodotto Azure equivalente per evitare la chiusura brusca dei carichi di lavoro.

Importante

Questa funzionalità è attualmente in anteprima. Le condizioni supplementari per l'utilizzo per le anteprime di Microsoft Azure includono termini legali più validi applicabili alle funzionalità di Azure in versione beta, in anteprima o altrimenti non ancora rilasciate nella disponibilità generale. Per informazioni su questa anteprima specifica, vedere informazioni sull'anteprima di Azure HDInsight nel servizio Azure Kubernetes. Per domande o suggerimenti sulle funzionalità, inviare una richiesta in AskHDInsight con i dettagli e segui Microsoft per ulteriori aggiornamenti su Azure HDInsight Community.

In questo articolo è possibile apprendere come arricchire gli eventi in tempo reale unendo un flusso da Kafka con una tabella in ADLS Gen2 usando Flink Streaming. L'API Flink Streaming viene usata per unire eventi da HDInsight Kafka con attributi di ADLS Gen2. Inoltre, utilizziamo eventi uniti per attributi per registrare in un altro topic Kafka.

Prerequisiti

cluster Flink su HDInsight su AKS
Kafka cluster in HDInsight
- Assicurarsi che le impostazioni di rete siano prese in considerazione come descritto in Uso di Kafka in HDInsight per assicurarsi che HDInsight su AKS e i cluster HDInsight si trovino nella stessa VNet.
Per questa dimostrazione, stiamo utilizzando una macchina virtuale Windows come ambiente di sviluppo del progetto Maven nella stessa rete virtuale (VNet) di HDInsight su Azure Kubernetes Service (AKS).

Preparazione dell'argomento Kafka

Verrà creato un argomento denominato user_events.

Lo scopo è leggere un flusso di eventi in tempo reale da un argomento Kafka usando Flink. Abbiamo ogni evento con i seguenti campi:
```
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

Preparare il file in ADLS Gen2

Si sta creando un file denominato item attributes nella risorsa di archiviazione

Lo scopo è leggere un batch di item attributes da un file in ADLS Gen2. Ogni elemento include i campi seguenti:
```
item_id, 
brand, 
category, 
timestamp, 
```

Sviluppare il processo Apache Flink

In questo passaggio vengono eseguite le attività seguenti

Arricchire l'argomento user_events da Kafka aggiungendo item attributes da un file in ADLS Gen2.
Spingiamo il risultato di questo passaggio, come un'attività utente arricchita di eventi, in un topic di Kafka.

Sviluppare un progetto Maven

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>

Unisci l'argomento Kafka con il file ADLS Gen2

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;
            }
        }
    }
}

Impacchettare il file JAR e inoltrarlo ad Apache Flink

Il file JAR in pacchetto viene inviato a Flink:

Generare un topic `user_events` in tempo reale su Kafka

È possibile produrre eventi di comportamento utente in tempo reale user_events in Kafka.

Utilizzare il join di `itemAttributes` con `user_events` in Kafka

Ora stiamo utilizzando itemAttributes sugli eventi di attività utente relativi all'unione nel file system user_events.

Continuiamo a produrre e consumare l'attività utente e gli attributi degli elementi nelle immagini seguenti.

Riferimento

esempi di Flink
sito Web Apache Flink
Apache, Apache Kafka, Kafka, Apache Flink, Flink e i nomi associati dei progetti open source sono marchi di Apache Software Foundation (ASF).

Condividi tramite

Arricchire gli eventi di Apache Kafka® con attributi di ADLS Gen2 con Apache Flink®

Prerequisiti

Preparazione dell'argomento Kafka

Preparare il file in ADLS Gen2

Sviluppare il processo Apache Flink

Sviluppare un progetto Maven

Impacchettare il file JAR e inoltrarlo ad Apache Flink

Generare un topic `user_events` in tempo reale su Kafka

Utilizzare il join di `itemAttributes` con `user_events` in Kafka

Riferimento

Risorse aggiuntive

Condividi tramite

Arricchire gli eventi di Apache Kafka® con attributi di ADLS Gen2 con Apache Flink®

Prerequisiti

Preparazione dell'argomento Kafka

Preparare il file in ADLS Gen2

Sviluppare il processo Apache Flink

Sviluppare un progetto Maven

Impacchettare il file JAR e inoltrarlo ad Apache Flink

Generare un topic user_events in tempo reale su Kafka

Utilizzare il join di itemAttributes con user_events in Kafka

Riferimento

Risorse aggiuntive

Generare un topic `user_events` in tempo reale su Kafka

Utilizzare il join di `itemAttributes` con `user_events` in Kafka