使用 Kafka 主题流和 Cosmos DB 进行数据丰富

2025 年 5 月 15 日 | 阅读 5 分钟

引言

数据丰富是指通过与额外数据集进行连接来增强原始流数据的过程，以提供更多上下文并使其更有价值。在实时数据管道中，可以使用 Apache Kafka Streams 在将数据持久化到目标数据库（如 Azure Cosmos DB）之前，对其进行处理、转换和丰富。

Kafka Streams + Cosmos DB 数据丰富架构概述

使用 Kafka Streams 和 Cosmos DB 的典型数据丰富管道由多个组件组成

架构组件

Kafka Producer：将原始事件（例如，用户操作、日志、交易）流式传输到 Kafka 主题。
Kafka Streams 应用程序
- 从 Kafka 消费原始事件。
- 通过与外部数据（来自 Cosmos DB）连接来丰富事件。
- 将丰富后的数据发布到另一个 Kafka 主题。
Kafka Consumer：读取丰富后的数据以进行进一步处理。
Cosmos DB：存储丰富后的数据以供实时查询和分析。
Kafka Connect Sink Connector：将丰富后的数据从 Kafka 流式传输到 Cosmos DB。

设置环境

在实施架构之前，请确保已安装并配置以下内容

Apache Kafka（带 Zookeeper）
Azure Cosmos DB（NoSQL 数据库）
Kafka Connect 和 Cosmos DB Sink Connector
Java 和 Kafka Streams API
用于 Kafka Producer/Consumer 的 Spring Boot

实现 Kafka Producer（原始数据流式传输）

Kafka producer 会将原始事件（例如，用户交互）流式传输出去。

示例原始数据

{
  "userId": "U123",
  "songId": "S456",
  "timestamp": "2025-02-27T10:15:30Z"
}   

Kafka Producer 的 Java 代码

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import com.fasterxml.jackson.databind.ObjectMapper;
public class KafkaRawEventProducer {
    private static final String TOPIC = "raw-events";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<>(props);
        ObjectMapper objectMapper = new ObjectMapper();

        try {
            RawEvent event = new RawEvent("U123", "S456", "2025-02-27T10:15:30Z");
            String message = objectMapper.writeValueAsString(event);

            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, event.getUserId(), message);
            producer.send(record, (metadata, exception) -> {
                if (exception == null) {
                    System.out.println("Produced event: " + message);
                } else {
                    exception.printStackTrace();
                }
            });

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }

    static class RawEvent {
        public String userId;
        public String songId;
        public String timestamp;

        public RawEvent(String userId, String songId, String timestamp) {
            this.userId = userId;
            this.songId = songId;
            this.timestamp = timestamp;
        }

        public String getUserId() { return userId; }
    }
}   

输出

Data Enrichment with Kafka Topics Streams and Cosmos DB

实现 Kafka Streams（数据丰富）

Kafka Streams 从 Cosmos DB 获取外部元数据并丰富事件。

Cosmos DB 中的外部元数据

Cosmos DB 存储附加数据，例如，歌曲元数据

{
  "songId": "S456",
  "title": "Afrobeat Groove",
  "artist": "Tony Allen"
}   

Kafka Streams Processor 的 Java 代码

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

import java.util.Properties;

public class KafkaDataEnrichmentProcessor {
    private static final String INPUT_TOPIC = "raw-events";
    private static final String OUTPUT_TOPIC = "enriched-events";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "data-enrichment-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        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<String, String> rawEvents = builder.stream(INPUT_TOPIC);

        KTable<String, String> songMetadata = builder.table("song-metadata", Consumed.with(Serdes.String(), Serdes.String()));

        KStream<String, String> enrichedStream = rawEvents.join(
            songMetadata,
            (rawValue, metadata) -> {
                if (metadata != null) {
                    return rawValue.replace("}", ", \"metadata\":" + metadata + "}");
                }
                return rawValue;
            },
            Joined.with(Serdes.String(), Serdes.String(), Serdes.String())
        );

        enrichedStream.to(OUTPUT_TOPIC);
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}   

输出

实现 Kafka Consumer

消费丰富后的数据以进行进一步处理。

Kafka Consumer 的 Java 代码

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaEnrichedConsumer {
    private static final String TOPIC = "enriched-events";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String GROUP_ID = "consumer-group";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(TOPIC));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Consumed enriched event: " + record.value());
            }
        }
    }
}   

输出

将丰富后的数据流式传输到 Cosmos DB

使用 Kafka Connect 和 Cosmos DB Sink Connector

{
  "name": "cosmosdb-sink",
  "config": {
    "connector.class": "com.azure.cosmos.kafka.connect.sink.CosmosDBSinkConnector",
    "tasks.max": "1",
    "topics": "enriched-events",
    "cosmos.account.endpoint": "https://your-cosmos-account.documents.azure.com:443/",
    "cosmos.account.key": "your-key",
    "cosmos.database.name": "musicDB",
    "cosmos.container.name": "enrichedEvents"
  }
}   

用于数据丰富的 Kafka Streams

Kafka Streams 是一个强大的流处理库，它允许我们通过将实时事件与静态或动态数据集连接来丰富流数据。

用例示例

想象一个音乐流媒体服务，其中原始的歌曲播放事件是实时生成的。然而，原始事件只包含歌曲 ID，我们需要在将它们存储到 Cosmos DB 之前，用歌曲元数据（例如艺术家姓名、流派和专辑）来丰富它们。

原始事件（Kafka Topic: song-plays）

丰富数据（Kafka Topic: song-metadata）

{"songId": "S456", "artist": "Tony Allen", "genre": "Afrobeat", "album": "Secret Agent"}   

丰富后的数据输出（Kafka Topic: enriched-song-plays）

Kafka Streams 将实时连接 song-plays 主题和song-metadata 主题，并生成一个丰富后的事件。

设置 Kafka Topics

我们需要三个主题

song-plays：存储原始歌曲播放事件。
song-metadata：存储歌曲元数据。
enriched-song-plays：存储丰富后的歌曲播放事件。

使用 Kafka CLI 创建这些主题

kafka-topics.sh --create --topic song-plays --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1
kafka-topics.sh --create --topic enriched-song-plays --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1   

用于原始歌曲播放事件的 Kafka Producer

以下 Python producer 将原始歌曲播放事件发送到 song-plays 主题。

song_plays_producer.py

from kafka import KafkaProducer
import json
import time
import random

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

songs = ["S456", "S789", "S101"]
users = ["U123", "U456", "U789"]

while True:
    event = {
        "userId": random.choice(users),
        "songId": random.choice(songs),
        "timestamp": int(time.time())
    }
    producer.send("song-plays", event)
    print(f"Produced: {event}")
    time.sleep(2)   

输出

用于歌曲元数据的 Kafka Producer

此 producer 将歌曲元数据发送到 song-metadata 主题。

song_metadata_producer.py

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

metadata = [
    {"songId": "S456", "artist": "Tony Allen", "genre": "Afrobeat", "album": "Secret Agent"},
    {"songId": "S789", "artist": "Fela Kuti", "genre": "Afrobeat", "album": "Zombie"},
    {"songId": "S101", "artist": "Burna Boy", "genre": "Afrobeats", "album": "African Giant"}
]

for song in metadata:
    producer.send("song-metadata", song)
    print(f"Produced: {song}")   

输出

实现用于数据丰富功能的 Kafka Streams

我们在 Java 中使用 Kafka Streams 连接 song-plays 主题和 song-metadata 主题。

KafkaStreamsEnrichment.java

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
import java.util.Properties;

public class KafkaStreamsEnrichment {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "song-enrichment-app");
        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();

        KTable<String, String> songMetadata = builder.table("song-metadata");
        KStream<String, String> songPlays = builder.stream("song-plays");

        KStream<String, String> enrichedPlays = songPlays
                .join(songMetadata, (play, metadata) -> play + ", " + metadata);

        enrichedPlays.to("enriched-song-plays");

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

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}   

用于将数据存储到 Cosmos DB 的 Kafka Consumer

我们使用 Python 消费丰富后的数据并将其存储到 Cosmos DB。

cosmosdb_consumer.py

from kafka import KafkaConsumer
import json
from azure.cosmos import CosmosClient

cosmos_url = "https://your-cosmosdb.documents.azure.com:443/"
cosmos_key = "your-primary-key"
database_name = "musicDB"
container_name = "enrichedSongs"

client = CosmosClient(cosmos_url, cosmos_key)
database = client.create_database_if_not_exists(database_name)
container = database.create_container_if_not_exists(container_name, partition_key="/songId")

consumer = KafkaConsumer(
    "enriched-song-plays",
    bootstrap_servers="localhost:9092",
    value_deserializer=lambda v: json.loads(v.decode("utf-8"))
)

for message in consumer:
    container.create_item(body=message.value)
    print(f"Stored in Cosmos DB: {message.value}")   

输出

下一个主题动态主题分区策略

使用 Kafka 主题流和 Cosmos DB 进行数据丰富

引言

Kafka Streams + Cosmos DB 数据丰富架构概述

架构组件

设置环境

实现 Kafka Producer（原始数据流式传输）

实现 Kafka Streams（数据丰富）

实现 Kafka Consumer

将丰富后的数据流式传输到 Cosmos DB

用于数据丰富的 Kafka Streams

用例示例

设置 Kafka Topics

用于原始歌曲播放事件的 Kafka Producer

用于歌曲元数据的 Kafka Producer

实现用于数据丰富功能的 Kafka Streams

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Kafka 教程

Kafka 安装

Kafka CLI

Kafka 编程

实时示例

Kafka 监控

Kafka Connect

Kafka Streams

杂项

使用 Kafka 主题流和 Cosmos DB 进行数据丰富

引言

Kafka Streams + Cosmos DB 数据丰富架构概述

架构组件

设置环境

实现 Kafka Producer（原始数据流式传输）

实现 Kafka Streams（数据丰富）

实现 Kafka Consumer

将丰富后的数据流式传输到 Cosmos DB

用于数据丰富的 Kafka Streams

用例示例

设置 Kafka Topics

用于原始歌曲播放事件的 Kafka Producer

用于歌曲元数据的 Kafka Producer

实现用于数据丰富功能的 Kafka Streams

相关帖子

Kafka 日志压缩

从 Kafka 到 Cosmos DB 的变更数据捕获 (CDC)

利用 Kafka Connect 进行 Cosmos DB 集成

使用 Kafka 进行事件驱动的供应链优化

Kafka Streams 的时间旅行查询

使用 Kafka、Azure 函数和 Cosmos DB 进行端到端数据流

使用 Kafka 进行零停机升级

Kafka 日志分析

通过 Kafka 实现量子实时数据流

使用 Kafka 向 Cosmos DB 进行实时流式传输

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器