Apache Flink® で ADLS Gen2 の属性を使用して Apache Kafka® のイベントを強化する

[アーティクル]
10/24/2024

Note

Azure HDInsight on AKS は 2025 年 1 月 31 日に廃止されます。 2025 年 1 月 31 日より前に、ワークロードを Microsoft Fabric または同等の Azure 製品に移行することで、ワークロードの突然の終了を回避する必要があります。サブスクリプション上に残っているクラスターは停止され、ホストから削除されることになります。

提供終了日までは基本サポートのみが利用できます。

重要

現在、この機能はプレビュー段階にあります。ベータ版、プレビュー版、または一般提供としてまだリリースされていない Azure の機能に適用されるその他の法律条項については、「Microsoft Azure プレビューの追加の使用条件」に記載されています。この特定のプレビューについては、「Microsoft 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 に確実に配置するためです
このデモでは、Windows 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 Web サイト
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` を使用する