使用kafka streams、KStream-GlobalKtable连接进行数据丰富

基础概念

Kafka Streams 是一个用于构建流处理应用程序和微服务的客户端库。它允许你在 Kafka 集群上进行实时数据处理，无需单独的处理集群。KStream 和 GlobalKTable 是 Kafka Streams 中的两种主要组件：

KStream：表示一个持续的数据流，可以理解为一个不断变化的数据表。它支持各种转换操作，如过滤、映射、聚合等。
GlobalKTable：表示一个全局的键值对表，数据会被缓存到内存中，并且支持高效的查找和更新操作。

类型

Kafka Streams 提供了多种类型的数据处理操作，包括：

转换操作：如 map、filter、flatMap 等。
聚合操作：如 groupByKey、reduce、aggregate 等。
连接操作：如 join、leftJoin、outerJoin 等。

应用场景

Kafka Streams 适用于多种实时数据处理场景，例如：

日志处理：实时处理和分析日志数据。
事件处理：实时处理和分析事件流数据。
数据集成：将来自不同数据源的数据进行实时集成和处理。
实时分析：对实时数据流进行聚合和分析。

连接 KStream 和 GlobalKTable 进行数据丰富

假设我们有一个 KStream，表示用户的行为数据，另一个 GlobalKTable，表示用户的详细信息。我们可以通过连接这两个组件来丰富用户的行为数据。

示例代码

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.GlobalKTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;

import java.util.Properties;

public class DataEnrichmentExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "data-enrichment-example");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();

        // 创建 KStream，表示用户的行为数据
        KStream<String, String> userActions = builder.stream("user-actions");

        // 创建 GlobalKTable，表示用户的详细信息
        GlobalKTable<String, String> userDetails = builder.globalTable("user-details", Materialized.as("user-details-store"));

        // 连接 KStream 和 GlobalKTable 进行数据丰富
        KStream<String, String> enrichedUserActions = userActions.join(
            userDetails,
            (actionValue, detailValue) -> actionValue + " - " + detailValue
        );

        // 输出丰富后的数据到新的主题
        enrichedUserActions.to("enriched-user-actions", Produced.with(Serdes.String(), Serdes.String()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}