Apache Flink® ile ADLS 2. Nesil öznitelikleriyle Apache Kafka'dan® olayları zenginleştirme

Makale
04/16/2024

Önemli

Bu özellik şu anda önizlemededir. Microsoft Azure Önizlemeleri için Ek Kullanım Koşulları, beta, önizleme aşamasında olan veya henüz genel kullanıma sunulmamış Azure özellikleri için geçerli olan daha fazla yasal hüküm içerir. Bu belirli önizleme hakkında bilgi için bkz . AKS üzerinde Azure HDInsight önizleme bilgileri. Sorular veya özellik önerileri için lütfen AskHDInsight'ta ayrıntıları içeren bir istek gönderin ve Azure HDInsight Topluluğu hakkında daha fazla güncelleştirme için bizi takip edin.

Bu makalede, Flink Streaming kullanarak ADLS 2. Nesil'deki tabloyla Kafka'dan bir akışa katılarak gerçek zamanlı olayları nasıl zenginleştirebileceğinizi öğrenebilirsiniz. HDInsight Kafka'dan gelen olayları ADLS 2. Nesil öznitelikleriyle birleştirmek için Flink Akış API'sini kullanırız. Ayrıca özniteliklere katılmış olayları başka bir Kafka konusuna eklemek için kullanırız.

Önkoşullar

AKS üzerinde HDInsight üzerinde Flink kümesi
HDInsight üzerinde Kafka kümesi
- AKS ve HDInsight kümelerinde HDInsight'ın aynı sanal ağda olduğundan emin olmak için HDInsight'ta Kafka kullanma konusunda açıklandığı gibi ağ ayarlarının dikkate alındığından emin olun
Bu tanıtım için, AKS üzerinde HDInsight ile aynı sanal ağda maven projesi geliştirme ortamı olarak bir Window VM kullanıyoruz

Kafka konu hazırlığı

adlı user_eventsbir konu oluşturuyoruz.

Amaç, Flink kullanarak kafka konusundan gerçek zamanlı olayların akışını okumaktır. Aşağıdaki alanlara sahip her olayımız var:
```
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

ADLS 2. Nesil'de dosya hazırlama

Depolama alanımızda adlı item attributes bir dosya oluşturuyoruz

Amaç, ADLS 2. Nesil'de bir dosyadan toplu iş item attributes okumaktır. Her öğe aşağıdaki alanlara sahiptir:
```
item_id, 
brand, 
category, 
timestamp, 
```

Apache Flink işini geliştirme

Bu adımda aşağıdaki etkinlikleri gerçekleştireceğiz

user_events ADLS 2. Nesil'de bir dosyadan ile item attributes katılarak Kafka'dan konuyu zenginleştirin.
Olayların zenginleştirilmiş kullanıcı etkinliği olarak bu adımın sonucunu kafka konusuna göndeririz.

Maven projesi geliştirme

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>

KAFKA konusuna ADLS 2. Nesil Dosyası ile katılma

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

Paket jar ve Apache Flink'e gönderme

Paketlenmiş jar dosyasını Flink'e gönderiyoruz:

Kafka'da gerçek zamanlı `user_events` konu oluşturma

Kafka'da gerçek zamanlı kullanıcı davranışı olayı user_events üretebiliyoruz.

Birleştirmeyi Kafka'da `itemAttributes` kullanma `user_events`

Artık dosya sisteminde kullanıcı etkinliği olaylarına user_eventskatıl 'i kullanıyoruzitemAttributes.

Aşağıdaki görüntülerde kullanıcı etkinliği ve öğe özniteliklerini üretmeye ve tüketmeye devam ediyoruz

Başvuru

Flink Örnekleri
Apache Flink Web Sitesi
Apache, Apache Kafka, Kafka, Apache Flink, Flink ve ilişkili açık kaynak proje adları Apache Software Foundation'ın (ASF) ticari markalarıdır.

Aracılığıyla paylaş

Apache Flink® ile ADLS 2. Nesil öznitelikleriyle Apache Kafka'dan® olayları zenginleştirme

Önkoşullar

Kafka konu hazırlığı

ADLS 2. Nesil'de dosya hazırlama

Apache Flink işini geliştirme

Maven projesi geliştirme

Paket jar ve Apache Flink'e gönderme

Kafka'da gerçek zamanlı `user_events` konu oluşturma

Birleştirmeyi Kafka'da `itemAttributes` kullanma `user_events`

Başvuru

Geri Bildirim

Geri Bildirim

Ek kaynaklar

Aracılığıyla paylaş

Apache Flink® ile ADLS 2. Nesil öznitelikleriyle Apache Kafka'dan® olayları zenginleştirme

Önkoşullar

Kafka konu hazırlığı

ADLS 2. Nesil'de dosya hazırlama

Apache Flink işini geliştirme

Maven projesi geliştirme

Paket jar ve Apache Flink'e gönderme

Kafka'da gerçek zamanlı user_events konu oluşturma

Birleştirmeyi Kafka'da itemAttributes kullanma user_events

Başvuru

Geri Bildirim

Geri Bildirim

Ek kaynaklar

Kafka'da gerçek zamanlı `user_events` konu oluşturma

Birleştirmeyi Kafka'da `itemAttributes` kullanma `user_events`