Obohacení událostí z Apache Kafka® o atributy z ADLS Gen2 pomocí Apache Flinku®

Článek
09/20/2024

Důležitý

Azure HDInsight v AKS byl vyřazen 31. ledna 2025. Další informace z tohoto oznámení.

Abyste se vyhnuli náhlému ukončení úloh, musíte migrovat úlohy do Microsoft Fabric nebo ekvivalentního produktu Azure.

Důležitý

Tato funkce je aktuálně ve verzi Preview. doplňkové podmínky použití pro verze Preview Microsoft Azure obsahují další právní podmínky, které se vztahují na funkce Azure, které jsou v beta verzi, ve verzi Preview nebo jinak ještě nebyly vydány k obecnému použití. Informace o této konkrétní verzi Preview najdete v tématu Azure HDInsight na AKS. Pokud máte dotazy nebo návrhy funkcí, odešlete prosím žádost na AskHDInsight s podrobnostmi a sledujte nás pro další aktualizace v komunitě Azure HDInsight na .

V tomto článku se dozvíte, jak můžete rozšířit události v reálném čase tím, že připojíte stream z Kafka s tabulkou v ADLS Gen2 pomocí Flink Streamingu. K připojení událostí z HDInsight Kafka s atributy z ADLS Gen2 používáme rozhraní API pro streamování Flink. Dále používáme události sloučené podle atributů, které posíláme do dalšího Kafka tématu.

Požadavky

Flink cluster ve službě HDInsight na AKS
Kafka cluster ve službě HDInsight
- Ujistěte se, že jsou nastavení sítě zajištěna podle popisu v Použití systému Kafka ve službě HDInsight, abyste zajistili, že HDInsight na AKS a clustery HDInsight jsou ve stejné virtuální síti (VNet).
Pro tuto ukázku používáme virtuální počítač s Windows jako vývojové prostředí pro projekty Maven ve stejné virtuální síti jako HDInsight v Azure Kubernetes Service (AKS).

Příprava témat pro Kafka

Vytváříme téma s názvem user_events.

Účelem je čtení datového proudu událostí v reálném čase z tématu Kafka pomocí Flinku. Každá událost má následující pole:
```
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

Příprava souboru v ADLS Gen2

Vytváříme soubor s názvem item attributes v našem úložišti.

Účelem je přečíst dávku item attributes ze souboru v ADLS Gen2. Každá položka má následující pole:
```
item_id, 
brand, 
category, 
timestamp, 
```

Vývoj úlohy Apache Flink

V tomto kroku provedeme následující aktivity:

Rozšíření tématu user_events ze systému Kafka připojením k item attributes ze souboru v ADLS Gen2
Výsledek tohoto kroku odesíláme jako rozšířenou uživatelskou aktivitu událostí do jednoho tématu Kafka.

Vývoj projektu 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>

Spojit téma Kafka se souborem 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;
            }
        }
    }
}

Zabalte soubor JAR a odešlete ho do Apache Flinku.

Balíček jar odesíláme do Flinku:

Vytváření tématu `user_events` v reálném čase na platformě Kafka

V systému Kafka můžeme vytvářet události chování uživatelů v reálném čase user_events.

Využití připojení `itemAttributes` s `user_events` v systému Kafka

Nyní používáme itemAttributes v systému souborů připojit události aktivity uživatele user_events.

V následujících obrázcích pokračujeme v vytváření a využívání atributů aktivity uživatelů a položek.

Odkaz

příklady Flinku
Webové stránky Apache Flink
Názvy projektů Apache, Apache Kafka, Kafka, Apache Flink, Flink a přidružených open source projektů jsou ochranné známky Apache Software Foundation (ASF).

Sdílet prostřednictvím

Obohacení událostí z Apache Kafka® o atributy z ADLS Gen2 pomocí Apache Flinku®

Požadavky

Příprava témat pro Kafka

Příprava souboru v ADLS Gen2

Vývoj úlohy Apache Flink

Vývoj projektu Maven

Zabalte soubor JAR a odešlete ho do Apache Flinku.

Vytváření tématu `user_events` v reálném čase na platformě Kafka

Využití připojení `itemAttributes` s `user_events` v systému Kafka

Odkaz

Další materiály

Sdílet prostřednictvím

Obohacení událostí z Apache Kafka® o atributy z ADLS Gen2 pomocí Apache Flinku®

Požadavky

Příprava témat pro Kafka

Příprava souboru v ADLS Gen2

Vývoj úlohy Apache Flink

Vývoj projektu Maven

Zabalte soubor JAR a odešlete ho do Apache Flinku.

Vytváření tématu user_events v reálném čase na platformě Kafka

Využití připojení itemAttributes s user_events v systému Kafka

Odkaz

Další materiály

Vytváření tématu `user_events` v reálném čase na platformě Kafka

Využití připojení `itemAttributes` s `user_events` v systému Kafka