使用 Kafka 和 Cosmos DB 构建可扩展的管道

2025 年 5 月 16 日 | 阅读 8 分钟

当今世界，处理和理解大量实时数据至关重要。创建快速且可扩展的数据管道有助于应用程序满足现代任务的需求。Apache Kafka 和 Azure Cosmos DB 是两个功能强大的工具，它们协同工作，构建可靠的数据管道。

1. Kafka 和 Cosmos DB 简介

Apache Kafka 概述

Apache Kafka 是一个分布式事件流平台，用于

实时数据处理： 每秒处理数百万条消息。
可扩展性： 无缝添加生产者、代理或消费者。
持久性： 具有复制功能的持久化日志存储，提供容错能力。

核心概念

生产者： 将消息发布到 Kafka 主题。
消费者： 从主题读取消息。
代理： 管理消息分发和存储。
主题： 消息的类别或馈送名称。

Azure Cosmos DB 概述

Azure Cosmos DB 专为高可用性、可扩展性和低延迟而设计。

主要特点

多模型支持： 包括文档、图形、键值和列族数据模型。
全局分发： 自动将数据复制到 Azure 各个区域。
一致性模型： 提供强一致性、最终一致性和有界陈旧性一致性。
完全托管服务： 处理缩放、备份和维护。

为何结合 Kafka 和 Cosmos DB？

这种组合提供了

实时数据管道： 使用 Kafka 摄取数据流并将其传输到 Cosmos DB。
可扩展的存储和查询： Cosmos DB 提供对流式数据的快速查询。
弹性： Kafka 的容错能力与 Cosmos DB 的全局分发相辅相成。

程序示例：Kafka 生产者到 Cosmos DB

以下是如何使用 Kafka 将实时流式数据发送到 Cosmos DB 的示例。

步骤 1：Kafka 生产者

Kafka 生产者将用户活动事件发布到某个主题。

from kafka import KafkaProducer
import json
import time

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)
user_activity = [
    {"user_id": "1", "action": "play", "song_id": "101", "timestamp": "2025-01-20T10:00:00Z"},
    {"user_id": "2", "action": "pause", "song_id": "102", "timestamp": "2025-01-20T10:05:00Z"}
]

# Sending Data to Kafka Topic
for activity in user_activity:
    producer.send('user_activity', value=activity)
    print(f"Sent: {activity}")
    time.sleep(1)

producer.flush()
producer.close()   

输出

Building Scalable Pipelines with Kafka and Cosmos DB

步骤 2：Kafka 消费者到 Cosmos DB

Kafka 消费者读取消息并将其插入 Cosmos DB。

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

# Cosmos DB Configuration
cosmos_client = CosmosClient("https://<your-account-name>.documents.azure.com:443/", "<your-primary-key>")
database_name = "KafkaDemo"
container_name = "UserActivity"

# Create Database and Container if they don't exist
database = cosmos_client.create_database_if_not_exists(database_name)
container = database.create_container_if_not_exists(
    id=container_name,
    partition_key=PartitionKey(path="/user_id")
)

# Kafka Consumer Configuration
consumer = KafkaConsumer(
    'user_activity',
    bootstrap_servers='localhost:9092',
    value_deserializer=lambda x: json.loads(x.decode('utf-8')),
    auto_offset_reset='earliest',
    group_id='cosmos_db_group'
)
for message in consumer:
    activity = message.value
    container.upsert_item(activity)
    print(f"Inserted into Cosmos DB: {activity}")   

输出

可扩展数据管道的架构

可扩展数据管道的架构旨在处理大容量、实时和容错处理，确保数据以最小的延迟高效地流经各个阶段。

架构概述

数据摄入
- 生产者将数据推送到 Kafka 主题。这些可以是实时数据源，如传感器、日志或来自 Web 应用程序的事件驱动数据。
- Kafka 作为一种持久的分布式消息代理，它缓冲数据并确保无数据丢失。
Stream 处理
- Kafka 消费者订阅主题并将数据传递给流处理框架，如 Apache Spark Streaming。
- Spark 根据延迟要求，以微批次或连续方式处理数据。
数据存储
- 处理后的数据将推送到数据仓库，如 Google BigQuery，用于分析目的。
- 存储层设计为随着数据量的增长而水平扩展。
可视化与监控
- 使用 Looker Studio 或其他 BI 工具构建仪表板和实时可视化，以深入了解数据。使用 Kafka、Spark Streaming 和 Google BigQuery 构建可扩展数据管道

1. Kafka 生产者代码

此 Kafka 生产者模拟实时事件数据，用于流式传输到 Kafka。

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class RealTimeKafkaProducer {
    public static void main(String[] args) {
        // Kafka producer configuration
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092"); // Kafka broker address
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
                // Send event to Kafka topic
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, event);

        String topic = "events";

        try {
            for (int i = 0; i < 100; i++) {
                // Simulate real-time event data
                String event = String.format("{\"event_id\":%d,\"event_type\":\"click\",\"timestamp\":%d}", i, System.currentTimeMillis() / 1000);
                producer.send(record);

                System.out.printf("Event sent: %s%n", event);

                // Simulate delay for real-time interval
                TimeUnit.SECONDS.sleep(1);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}   

输出

2. Spark Streaming 消费者代码

此 Spark Streaming 消费者处理来自 Kafka 的数据，执行转换，并将结果存储在 BigQuery 中。

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;
import org.apache.spark.sql.functions;

public class KafkaToBigQuery {
    public static void main(String[] args) {        SparkSession spark = SparkSession.builder()
                .appName("KafkaSparkStreamingToBigQuery")
                .master("local[*]")  // Use all available cores
                .getOrCreate();

        // Read stream from Kafka topic
        Dataset<Row> kafkaStream = spark
                .readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", "localhost:9092")
                .option("subscribe", "events")  // Kafka topic name
                .load();
        Dataset<Row> jsonData = kafkaStream.selectExpr("CAST(value AS STRING) AS json_data");
        Dataset<Row> parsedData = jsonData.select(functions.from_json(
                functions.col("json_data"),
                spark.read().schema("{\"event_id\": INT, \"event_type\": STRING, \"timestamp\": LONG}")
        ).as("event"))
                .selectExpr("event.event_id", "event.event_type", "event.timestamp");
        try {
            StreamingQuery query = parsedData
                    .writeStream()
                    .format("bigquery")
                    .option("table", "your-project.your_dataset.events")  // BigQuery table
                    .option("checkpointLocation", "gs://your-bucket/checkpoint/")  // GCS checkpoint
                    .outputMode("append")
                    .start();

            query.awaitTermination();  // Keep the stream running
        } catch (StreamingQueryException e) {
            e.printStackTrace();
        }
    }
}   

3. 用于数据存储的 BigQuery Schema

数据由 Spark Streaming 处理后，它将被存储在具有以下 Schema 的 BigQuery 表中

event_id (INTEGER)
event_type (STRING)
timestamp (TIMESTAMP)

您需要确保 Spark 写入 Google BigQuery 的必要权限已设置。

4. 处理容错与可扩展性

Kafka 通过主题复制（默认 3 个副本）确保消息持久性。
Spark Streaming 通过利用检查点来处理精确一次语义。
要扩展管道，您可以增加 Kafka 分区、Spark 执行器或 BigQuery 的分区表。

输出

安装和配置 Kafka

步骤 1：安装 Apache Kafka

Download Kafka from the official website and extract the contents. This guide assumes Kafka is installed on a Unix-like system.
# Download Kafka
wget https://downloads.apache.org/kafka/<version>/kafka_<scala_version>-<version>.tgz
tar -xzf kafka_<scala_version>-<version>.tgz
cd kafka_<scala_version>-<version>   

步骤 2：启动 Zookeeper 服务器

Kafka 依赖 Zookeeper 来管理其集群元数据。使用提供的配置文件启动 ZooKeeper 服务。

# Start Zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties   

步骤 3：启动 Kafka Broker

Zookeeper 运行后，您可以启动 Kafka Broker。

# Start Kafka server
bin/kafka-server-start.sh config/server.properties   

步骤 4：验证 Kafka 是否正在运行

要检查 Kafka 是否正在运行，请使用 Kafka 服务器的 kafka-topics.sh 命令列出主题。

# List Kafka topics (should return no topics if none are created yet)
bin/kafka-topics.sh --list --bootstrap-server localhost:9092   

为数据管道设置主题

步骤 1：创建 Kafka 主题

Kafka 主题是数据管道的基本构建块。您可以使用以下命令创建主题。

# Create a new Kafka topic
bin/kafka-topics.sh --create --topic <your_topic_name> \
    --bootstrap-server localhost:9092 \
    --partitions 3 \
    --replication-factor 1   

此命令创建一个名为 <your_topic_name> 的主题，包含 3 个分区和 1 个复制因子。

步骤 2：验证主题创建

创建主题后，您可以通过再次列出主题来验证它。

# List all Kafka topics
bin/kafka-topics.sh --list --bootstrap-server localhost:9092   

步骤 3：设置主题配置（可选）

您可以使用以下命令配置主题的设置，如保留期或清理策略。

# Alter topic settings
bin/kafka-configs.sh --alter --entity-type topics --entity-name <your_topic_name> \
    --add-config retention.ms=604800000 --bootstrap-server localhost:9092   

此命令将保留期设置为 7 天（以毫秒为单位）。

步骤 4：向主题发布消息

您可以使用生产者开始向 Kafka 主题发送消息。

执行命令后，在终端中键入消息。它们将被发送到 Kafka 主题。

步骤 5：从主题消费消息

您可以通过消费消息来验证消息是否已发布到主题。

# Start a Kafka consumer to read messages from the topic
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic <your_topic_name> --from-beginning   

此命令将打印发送到该主题的所有消息，从头开始。

Cosmos DB 配置代码示例

使用 Azure CLI

# Create Cosmos DB Account
az cosmosdb create --name KafkaCosmosDB --resource-group MyResourceGroup

# Create Database
az cosmosdb sql database create \
    --account-name KafkaCosmosDB \
    --resource-group MyResourceGroup \
    --name KafkaPipelineDB

# Create Container
az cosmosdb sql container create \
    --account-name KafkaCosmosDB \
    --resource-group MyResourceGroup \
    --database-name KafkaPipelineDB \
    --name PipelineData \
    --partition-key-path /id   

5. 开发 Kafka 生产者

开发 Kafka 生产者

在 Kafka 中，生产者负责将记录（消息）发送到 Kafka 主题。让我们逐步完成开发 Kafka 生产者、编写 Kafka 生产者代码并运行它的过程。

1. 编写 Kafka 生产者代码

要编写 Kafka 生产者，您需要使用 Kafka 的生产者 API，该 API 可以用 Java、Python 等语言编写。以下是使用 Java 编写 Kafka 生产者的示例。

Java Kafka 生产者示例

1. 添加 Kafka 依赖项： 如果您使用 Maven，则需要在 pom.xml 中添加以下依赖项。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.0.0</version>
</dependency>   

2. 编写 Kafka 生产者代码

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class MyKafkaProducer {
    public static void main(String[] args) {
        // Set up producer properties
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // Create a KafkaProducer instance
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        // Create a ProducerRecord (message)
        String topic = "test-topic";
        String key = "key1";
        String value = "Hello Kafka!";
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

        // Send the record to Kafka
        try {
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    exception.printStackTrace();
                } else {
                    System.out.println("Message sent successfully to topic " + metadata.topic());
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}   

说明

servers：指定 Kafka Broker（此处为 localhost:9092）。
serializer 和 value.serializer：定义在发送之前如何序列化键和值。
KafkaProducer：与 Kafka Broker 交互的主要生产者类。
ProducerRecord：表示将要发送到主题的消息。

2. 运行 Kafka 生产者

编写完 Kafka 生产者代码后，您可以在开发环境中编译并运行它。以下是步骤：

启动 Kafka 服务器： 确保 Kafka 服务器已启动并正在运行。如果您使用的是本地设置，则可以通过执行以下命令来运行 Kafka 服务器：

启动 Zookeeper（如果尚未运行）

启动 Kafka Broker

创建 Kafka 主题： 在运行生产者之前，您需要创建 Kafka 主题（在本例中为 test-topic）。

bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1   

编译生产者代码： 如果您使用 Maven，可以使用以下命令进行编译：

运行 Kafka 生产者代码： 要运行生产者，请使用以下命令：

或者，如果您使用的是 IntelliJ IDEA 或 Eclipse 等 IDE，您可以直接运行 MyKafkaProducer 类。

预期输出

当生产者成功执行时，将显示以下输出：

您还可以检查日志以验证消息是否正在发布。为此，您可以运行 Kafka 消费者来验证数据。

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning   

这应该会显示生产者发送的消息，确认 Kafka 生产者工作正常。

6. 为 Cosmos DB 创建 Kafka 消费者

编写 Kafka 消费者代码

Kafka 消费者从 Kafka 读取数据并将其插入 Cosmos DB。

Code Example

import com.azure.cosmos.*;
import com.azure.cosmos.models.*;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;

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

public class KafkaToCosmos {
    public static void main(String[] args) {
        // Cosmos DB Configuration
        String cosmosEndpoint = "<COSMOS_DB_ENDPOINT>";
        String cosmosKey = "<COSMOS_DB_KEY>";
        String databaseName = "KafkaPipelineDB";
        String containerName = "PipelineData";

        // Initialize Cosmos DB Client
        CosmosClient cosmosClient = new CosmosClientBuilder()
                .endpoint(cosmosEndpoint)
                .key(cosmosKey)
                .consistencyLevel(ConsistencyLevel.EVENTUAL)
                .buildClient();
        CosmosContainer container = cosmosClient.getDatabase(databaseName).getContainer(containerName);

        // Kafka Consumer Configuration
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "cosmos-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("kafka-cosmos-pipeline"));

        // Consume and Insert Data
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    String jsonData = record.value();
                    container.upsertItem(jsonData);
                    System.out.println("Inserted: " + jsonData);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
            cosmosClient.close();
        }
    }
}   

输出

下一个主题Change-data-capture-from-kafka-to-cosmos-db

使用 Kafka 和 Cosmos DB 构建可扩展的管道

1. Kafka 和 Cosmos DB 简介

Apache Kafka 概述

核心概念

Azure Cosmos DB 概述

主要特点

为何结合 Kafka 和 Cosmos DB？

可扩展数据管道的架构

安装和配置 Kafka

步骤 1：安装 Apache Kafka

步骤 2：启动 Zookeeper 服务器

步骤 3：启动 Kafka Broker

步骤 4：验证 Kafka 是否正在运行

为数据管道设置主题

步骤 1：创建 Kafka 主题

步骤 2：验证主题创建

步骤 3：设置主题配置（可选）

步骤 4：向主题发布消息

步骤 5：从主题消费消息

Cosmos DB 配置代码示例

5. 开发 Kafka 生产者

开发 Kafka 生产者

6. 为 Cosmos DB 创建 Kafka 消费者

编写 Kafka 消费者代码

联系信息

关注我们

教程

面试题

在线编译器

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 构建可扩展的管道

1. Kafka 和 Cosmos DB 简介

Apache Kafka 概述

核心概念

Azure Cosmos DB 概述

主要特点

为何结合 Kafka 和 Cosmos DB？

可扩展数据管道的架构

安装和配置 Kafka

步骤 1：安装 Apache Kafka

步骤 2：启动 Zookeeper 服务器

步骤 3：启动 Kafka Broker

步骤 4：验证 Kafka 是否正在运行

为数据管道设置主题

步骤 1：创建 Kafka 主题

步骤 2：验证主题创建

步骤 3：设置主题配置（可选）

步骤 4：向主题发布消息

步骤 5：从主题消费消息

Cosmos DB 配置代码示例

5. 开发 Kafka 生产者

开发 Kafka 生产者

6. 为 Cosmos DB 创建 Kafka 消费者

编写 Kafka 消费者代码

相关帖子

使用 Kafka 实现不可变数据架构

Kafka 控制器

模式演进

使用 Kafka 进行事件驱动的网络安全

Kafka 中的日志压缩

Apache Kafka 幂等生产者

Kafka 中的系统监控和警报

Kafka 和 Cosmos DB 之间异步数据处理

BigQuery 模式映射的 Kafka 核心概念

Kafka 消息压缩

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器