使用來自 ADLS Gen2 與 Apache Flink® 的屬性擴充 Apache Kafka® 的事件

發行項
09/20/2024

重要

AKS 上的 Azure HDInsight 於 2025 年 1 月 31 日淘汰。透過此公告深入瞭解。

您必須將工作負載移轉至 Microsoft Fabric 或對等 Azure 產品，以避免突然終止工作負載。

重要

這項功能目前為預覽狀態。 Microsoft Azure 預覽版的補充使用規定包含適用於 Beta 版、預覽版或尚未正式發行之 Azure 功能的更合法條款。如需此特定預覽的相關資訊，請參閱 Azure HDInsight on AKS 預覽資訊。如有問題或功能建議，請至 AskHDInsight 提出請求，並關注我們以獲取 Azure HDInsight 社群的更多更新。

在本文中，您可以瞭解如何使用 Flink 串流，通過將 Kafka 的數據流與 ADLS Gen2 上的表格聯結來豐富即時事件。我們使用 Flink 串流 API 來聯結來自 HDInsight Kafka 的事件與 ADLS Gen2 中的屬性。此外，我們會使用已聯結屬性的事件來寫入另一個 Kafka 主題。

先決條件

AKS 上 HDInsight 上的 Flink 叢集
HDInsight 上的 Kafka 叢集
- 請確保已按照在 HDInsight 上使用 Kafka 的說明處理網路設定，確保 AKS 上的 HDInsight 和 HDInsight 叢集位於相同的 VNet。
在此示範中，我們會在與 AKS 上的 HDInsight 相同的 VNet 中使用 Window VM 作為 maven 專案開發環境

Kafka 主題準備

我們正在建立名為 user_events的主題。

目的是使用 Flink 從 Kafka 主題讀取即時事件的串流。我們有具有下列欄位的每個事件：
```
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 Gen2 上準備檔案

我們正在記憶體中建立名為 item attributes 的檔案

目的是從 ADLS Gen2 上的檔案讀取一批 item attributes。每個專案都有下列欄位：
```
item_id, 
brand, 
category, 
timestamp, 
```

開發 Apache Flink 任務

在此步驟中，我們會執行下列活動

使用來自 ADLS Gen2 上的檔案的 item attributes，以增強 Kafka 的 user_events 主題。
我們將這個步驟的結果作為豐富的用戶事件活動推送到 Kafka 主題。

開發 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>

使用 ADLS Gen2 檔案 加入 Kafka 主題

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

打包 jar，並提交至 Apache Flink

我們會將封裝的 jar 提交至 Flink：

在 Kafka 上產生即時 `user_events` 主題

我們能夠在 Kafka 中產生即時用戶行為事件 user_events。

在 Kafka 上，消耗 `itemAttributes` 與 `user_events` 的聯結

我們現在在文件系統上使用 itemAttributes，user_events加入用戶活動事件。

我們會繼續在下列影像中產生及取用用戶活動和項目屬性

參考

Flink 範例
Apache Flink 網站
Apache、Apache Kafka、Kafka、Apache Flink、Flink 和相關聯的開放原始碼專案名稱 Apache Software Foundation （ASF）商標。

共用方式為

使用來自 ADLS Gen2 與 Apache Flink® 的屬性擴充 Apache Kafka® 的事件

先決條件

Kafka 主題準備

在 ADLS Gen2 上準備檔案

開發 Apache Flink 任務

開發 Maven 專案

打包 jar，並提交至 Apache Flink

在 Kafka 上產生即時 `user_events` 主題

在 Kafka 上，消耗 `itemAttributes` 與 `user_events` 的聯結

參考

其他資源

共用方式為

使用來自 ADLS Gen2 與 Apache Flink® 的屬性擴充 Apache Kafka® 的事件

先決條件

Kafka 主題準備

在 ADLS Gen2 上準備檔案

開發 Apache Flink 任務

開發 Maven 專案

打包 jar，並提交至 Apache Flink

在 Kafka 上產生即時 user_events 主題

在 Kafka 上，消耗 itemAttributes 與 user_events 的聯結

參考

其他資源

在 Kafka 上產生即時 `user_events` 主題

在 Kafka 上，消耗 `itemAttributes` 與 `user_events` 的聯結