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

Artigo
09/20/2024

Importante

O Azure HDInsight no AKS se aposentou em 31 de janeiro de 2025. Saiba mais com este comunicado.

Você precisa migrar suas cargas de trabalho para microsoft fabric ou um produto equivalente do Azure para evitar o encerramento abrupto de suas cargas de trabalho.

Importante

Esse recurso está atualmente em versão prévia. Os termos de uso complementares para o Microsoft Azure Previews 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 visualização específica, consulte Azure HDInsight nas informações de visualização do AKS. Para perguntas ou sugestões de funcionalidades, envie uma solicitação no AskHDInsight com os detalhes e siga-nos para obter mais atualizações sobre a Comunidade Azure HDInsight .

Neste artigo, você pode aprender 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 eventos associados a atributos para afundar em outro tópico do Kafka.

Pré-requisitos

Cluster Flink no HDInsight no AKS
Cluster Kafka no HDInsight
- Verifique se as configurações de rede sejam configuradas conforme descrito em Usando o Kafka no HDInsight para garantir que o HDInsight no AKS e os clusters de HDInsight estejam na mesma VNet.
Para esta demonstração, estamos usando uma VM com Windows como o ambiente de desenvolvimento de projetos 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 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 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, executamos as seguintes atividades

Enriqueça o tópico user_events do Kafka combinando com item attributes a partir de um arquivo no ADLS Gen2.
Transmitimos o resultado desta etapa como uma atividade de usuário enriquecida por eventos para um tópico do Kafka.

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

Juntar o tópico 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;
            }
        }
    }
}

Empacote o jar e envie para o Apache Flink.

Estamos submetendo o jar empacotado para Flink.

Criar tópico `user_events` em tempo real no Kafka

Podemos produzir eventos de comportamento do usuário em tempo real no Kafka, como o evento user_events.

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

Agora estamos usando itemAttributes em eventos de adesão de atividades do usuário 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 software livre associados são marcas comerciais 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 arquivo no ADLS Gen2

Desenvolver o trabalho do Apache Flink

Desenvolver projeto do Maven

Empacote o jar e envie para o Apache Flink.

Criar tópico `user_events` em tempo real no Kafka

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

Referência

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 arquivo no ADLS Gen2

Desenvolver o trabalho do Apache Flink

Desenvolver projeto do Maven

Empacote o jar e envie para o Apache Flink.

Criar tópico user_events em tempo real no Kafka

Consumir a junção itemAttributes com user_events no Kafka

Referência

Recursos adicionais

Criar tópico `user_events` em tempo real no Kafka

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