Enriquecer os eventos do Apache Kafka® com atributos do ADLS Gen2 com o Apache Flink

Artigo
04/17/2024

Importante

Esse recurso está atualmente na visualização. Os Termos de Uso Complementares para Versões Prévias da Microsoft Azure incluem mais termos legais que se aplicam aos recursos do Azure que estão em versão beta, em versão prévia ou ainda não lançados em disponibilidade geral. Para obter informações sobre essa versão prévia específica, confira Informações sobre a versão prévia do HDInsight no AKS. No caso de perguntas ou sugestões de recursos, envie uma solicitação no AskHDInsight com os detalhes e siga-nos para ver mais atualizações sobre a Comunidade do Azure HDInsight.

Neste artigo, você saberá como enriquecer os eventos em tempo real unindo um fluxo do Kafka com tabela no ADLS Gen2 usando o Flink Streaming. Usamos a API de Streaming do Flink para unir eventos do HDInsight Kafka com atributos do ADLS Gen2. Além disso, usamos os eventos ingressados por atributos para coletar em outro tópico do Kafka.

Pré-requisitos

Cluster do Flink no HDInsight no AKS
Cluster do Kafka no HDInsight
- Certifique-se de que as configurações de rede sejam atendidas conforme descrito em Usando o Kafka no HDInsight para garantir que os clusters do HDInsight no AKS e do HDInsight estejam na mesma VNet
Para esta demonstração, estamos usando uma VM de ambiente de desenvolvimento de projeto maven na mesma VNet que o HDInsight no AKS

Preparação do tópico do Kafka

Estamos criando um tópico chamado user_events.

A finalidade é ler um fluxo de eventos em tempo real de um tópico do Kafka usando Flink. Temos todos os eventos com os seguintes campos:
```
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

Preparar o arquivo no ADLS Gen2

Estamos criando um arquivo chamado item attributes em nosso armazenamento

A finalidade é ler um lote de item attributes de um arquivo no ADLS Gen2. Cada item tem os seguintes campos:
```
item_id, 
brand, 
category, 
timestamp, 
```

Desenvolver o trabalho do Apache Flink

Nesta etapa, realizamos as seguintes atividades

Enriqueça o tópico user_events do Kafka unindo-se a item attributes de um arquivo no ADLS Gen2.
Enviamos por push o resultado dessa etapa, como uma atividade enriquecida do usuário de eventos em um tópico do Kafka.

Desenvolver um projeto 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>

Ingressar no tópico do Kafka com o arquivo 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;
            }
        }
    }
}

Empacotar o jar e enviar para o Apache Flink

Estamos enviando o jar empacotado para o Flink:

Produzir o tópico `user_events` em tempo real no Kafka

Conseguimos produzir o evento de comportamento do usuário em tempo real user_events no Kafka.

Consumir a junção `itemAttributes` com `user_events` no Kafka

Agora estamos usando itemAttributes nos eventos de atividade do usuário de ingresso no sistema de arquivos user_events.

Continuamos a produzir e consumir a atividade do usuário e os atributos de item nas imagens a seguir

Referência

Exemplos de Flink
Site do Apache Flink
Apache, Apache Kafka, Kafka, Apache Flink, Flink e nomes de projetos de código aberto associados são marcas registradas da Apache Software Foundation (ASF).

Compartilhar via

Enriquecer os eventos do Apache Kafka® com atributos do ADLS Gen2 com o Apache Flink

Pré-requisitos

Preparação do tópico do Kafka

Preparar o arquivo no ADLS Gen2

Desenvolver o trabalho do Apache Flink

Desenvolver um projeto Maven

Empacotar o jar e enviar para o Apache Flink

Produzir o tópico `user_events` em tempo real no Kafka

Consumir a junção `itemAttributes` com `user_events` no Kafka

Referência

Comentários

Comentários

Recursos adicionais

Compartilhar via

Enriquecer os eventos do Apache Kafka® com atributos do ADLS Gen2 com o Apache Flink

Pré-requisitos

Preparação do tópico do Kafka

Preparar o arquivo no ADLS Gen2

Desenvolver o trabalho do Apache Flink

Desenvolver um projeto Maven

Empacotar o jar e enviar para o Apache Flink

Produzir o tópico user_events em tempo real no Kafka

Consumir a junção itemAttributes com user_events no Kafka

Referência

Comentários

Comentários

Recursos adicionais

Produzir o tópico `user_events` em tempo real no Kafka

Consumir a junção `itemAttributes` com `user_events` no Kafka