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

發行項
10/23/2024

注意

AKS 上的 Azure HDInsight 將於 2025 年 1 月 31 日退場。請於 2025 年 1 月 31 日之前，將工作負載移轉至 Microsoft Fabric 或對等的 Azure 產品，以免工作負載突然終止。訂用帳戶中剩餘的叢集將會停止，並會從主機移除。

在淘汰日期之前，只有基本支援可用。

重要

此功能目前為預覽功能。 Microsoft Azure 預覽版增補使用規定包含適用於 Azure 功能 (搶鮮版 (Beta)、預覽版，或尚未正式發行的版本) 的更多法律條款。若需此特定預覽版的相關資訊，請參閱 Azure HDInsight on AKS 預覽版資訊。如有問題或功能建議，請在 AskHDInsight 上提交要求並附上詳細資料，並且在 Azure HDInsight 社群上追蹤我們以獲得更多更新資訊。

在本文中，您可以了解如何使用 Flink 串流，將 Kafka 的資料流與 ADLS Gen2 上的資料表聯結以擴充即時事件。我們使用 Flink Streaming API 聯結來自 HDInsight Kafka 的事件與來自 ADLS Gen2 的屬性。此外，我們使用已聯結屬性的事件來產生另一個 Kafka 主題。

必要條件

HDInsight on AKS 上的 Flink 叢集
HDInsight 上的 Kafka 叢集
- 確定網路設定已如使用 HDInsight 上的 Kafka 中所述進行處理；以確保 HDInsight on AKS 和 HDInsight 叢集位於相同的 VNet 中
在此示範中，我們會在與 HDInsight on AKS 相同的 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`