你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

通过 Apache Flink® 使用 ADLS Gen2 中的属性扩充 Apache Kafka® 中的事件

项目
10/16/2024

注意

我们将于 2025 年 1 月 31 日停用 Azure HDInsight on AKS。在 2025 年 1 月 31 日之前，你需要将工作负荷迁移到 Microsoft Fabric 或同等的 Azure 产品，以避免工作负荷突然终止。订阅上的剩余群集会被停止并从主机中移除。

在停用日期之前，仅提供基本支持。

重要

此功能目前以预览版提供。 Microsoft Azure 预览版的补充使用条款包含适用于 beta 版、预览版或其他尚未正式发布的 Azure 功能的更多法律条款。有关此特定预览版的信息，请参阅 Azure HDInsight on AKS 预览版信息。如有疑问或功能建议，请在 AskHDInsight 上提交请求并附上详细信息，并关注我们以获取 Azure HDInsight Community 的更多更新。

本文介绍如何使用 Flink 流式处理将 Kafka 中的流与 ADLS Gen2 上的表联接在一起，从而扩充实时事件。我们使用 Flink 流式处理 API 将 HDInsight Kafka 中的事件与 ADLS Gen2 中的属性联接。我们进一步使用已联接属性的事件接收器进入另一个 Kafka 主题。

先决条件

AKS 上的 HDInsight 上的 Flink 群集
HDInsight 上的 Kafka 群集
- 确保按照在 HDInsight 上使用 Kafka 中所述处理网络设置，以确保 AKS 和 HDInsight 群集上 HDInsight 位于同一 VNet 中
在本演示中，我们将使用 Window VM 作为 HDInsight on AKS 所在的同一 VNet 中的 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>

将 Kafka 主题与 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;
            }
        }
    }
}

打包 jar 并提交到 Apache Flink

我们将打包的 jar 提交到 Flink：

在 Kafka 上生成实时 `user_events` 主题

我们能够在 Kafka 中生成实时用户行为事件 user_events。

在 Kafka 上使用与 `user_events` 联接在一起的 `itemAttributes`

我们现在在文件系统联接用户活动事件 user_events 上使用 itemAttributes。

我们继续生成和使用下图中的用户活动和项属性

参考

Flink 示例
Apache Flink 网站
Apache、Apache Kafka、Kafka、Apache Flink、Flink 和关联的开源项目名称是 Apache Software Foundation (ASF) 的商标。

通过

通过 Apache Flink® 使用 ADLS Gen2 中的属性扩充 Apache Kafka® 中的事件

先决条件

Kafka 主题准备

在 ADLS Gen2 上准备文件

开发 Apache Flink 作业

开发 Maven 项目

打包 jar 并提交到 Apache Flink

在 Kafka 上生成实时 `user_events` 主题

在 Kafka 上使用与 `user_events` 联接在一起的 `itemAttributes`

参考

反馈

其他资源

通过

通过 Apache Flink® 使用 ADLS Gen2 中的属性扩充 Apache Kafka® 中的事件

先决条件

Kafka 主题准备

在 ADLS Gen2 上准备文件

开发 Apache Flink 作业

开发 Maven 项目

打包 jar 并提交到 Apache Flink

在 Kafka 上生成实时 user_events 主题

在 Kafka 上使用与 user_events 联接在一起的 itemAttributes

参考

反馈

其他资源

在 Kafka 上生成实时 `user_events` 主题

在 Kafka 上使用与 `user_events` 联接在一起的 `itemAttributes`