使用 Kafka 在 Cosmos DB 中处理大规模优化数据

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

引言

Apache Kafka 是一个分布式事件流平台，专为高吞吐量、容错性和实时处理而设计。它充当事件驱动型架构的中间件，能够实现事件生成者（事件生成器）和事件使用者（事件处理器）之间的数据无缝移动。另一方面，Azure Cosmos DB 是一种全球分布式 NoSQL 数据库服务，以其可伸缩性、低延迟性能和多模型能力而闻名。

通过将 Kafka 与 Cosmos DB 集成，组织可以建立一个高度可伸缩的实时数据处理管道，该管道能够高效地摄取、处理和持久化事件数据。

大规模事件处理的重要性

处理大规模事件数据对于各个行业都至关重要，包括

音乐流媒体服务：实时跟踪歌曲播放、用户交互和推荐。
电子商务：动态监控客户行为、购买和库存更新。
物联网和智能设备：收集来自数百万台连接设备的数据。
金融服务：实时检测欺诈、执行交易和分析市场趋势。
医疗保健：处理患者记录、监控医疗设备和跟踪健康指标。

处理这些工作负载需要一个能够

高吞吐量：每秒处理数百万个事件。
低延迟：确保实时响应和决策。
容错：防止发生故障时数据丢失。
可伸缩性：在不降低性能的情况下处理不断增长的数据量。
灵活的数据存储：高效支持结构化和非结构化数据。

Kafka 和 Cosmos DB 共同为这些需求提供了坚实的基础。

为什么将 Kafka 与 Cosmos DB 结合使用？

Kafka 和 Cosmos DB 通过提供实时事件处理和可伸缩存储来相互补充。它们的集成提供了多项优势

1. 可伸缩性和弹性

Kafka 的分区机制支持事件流的水平扩展。
Cosmos DB 的全球分布允许跨多个区域高效地存储和查询数据。

2. 高可用性和可靠性

Kafka 通过复制和分布式代理架构确保容错。
Cosmos DB 通过自动多区域故障转移保证 99.999% 的可用性。

3. 实时处理和分析

Kafka 允许使用消费者组进行实时事件流式传输和处理。
Cosmos DB 通过索引和分区提供快速查询性能。

4. 成本优化

Kafka 的事件保留和压缩降低了存储成本。
Cosmos DB 提供按需付费模式，优化资源利用率。

5. 多模型和灵活的模式支持

Kafka 支持 JSON、Avro 和 Protobuf 格式，以实现灵活的数据交换。
Cosmos DB 支持多种 API（SQL、MongoDB、Cassandra、Table、Gremlin），适用于不同的工作负载。

架构概述

Kafka 和 Cosmos DB 的集成涉及多个组件协同工作，以创建可伸缩且高效的数据管道。高层架构包括

事件生产者：生成实时事件的应用程序。
Kafka 集群：由多个管理事件流的代理组成。
Kafka 主题：在处理之前存储事件的逻辑通道。
Kafka 消费者组：读取事件并写入 Cosmos DB 的消费者。
Cosmos DB 存储：存储用于分析事件的分布式 NoSQL 数据库。

流程图

示例 Kafka 生产者代码

from kafka import KafkaProducer
import json

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

event = {'user': 'Alice', 'action': 'play', 'song': 'Imagine'}
producer.send('music-events', value=event)
producer.flush()
print("Event Sent:", event)

输出

Handling Large-scale Optimized Data in Cosmos DB Using Kafka

示例 Kafka 消费者代码

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'music-events',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    value_deserializer=lambda v: json.loads(v.decode('utf-8'))
)

for message in consumer:
    print("Received Event:", message.value)

输出

在 Cosmos DB 中存储事件

from azure.cosmos import CosmosClient

url = "<COSMOS_DB_ENDPOINT>"
key = "<COSMOS_DB_KEY>"
database_name = "MusicDB"
container_name = "Events"

client = CosmosClient(url, credential=key)
database = client.create_database_if_not_exists(id=database_name)
container = database.create_container_if_not_exists(id=container_name, partition_key='/user')
event = {'id': '1', 'user': 'Alice', 'action': 'play', 'song': 'Imagine'}
container.create_item(event)
print("Event Stored in Cosmos DB:", event)

输出

为大规模事件处理设置 Kafka

概述

为大规模事件处理设置 Kafka 包括安装 Kafka、配置代理、设置主题和优化性能。本节将介绍

安装和启动 Kafka。
配置 Kafka 以实现高吞吐量。
创建具有最佳配置的主题。
运行 Kafka 生产者和消费者。
为大规模数据处理进行性能调优。

步骤 1：安装和启动 Kafka

下载并解压 Kafka

wget https://downloads.apache.org/kafka/3.5.1/kafka_2.13-3.5.1.tgz
tar -xvzf kafka_2.13-3.5.1.tgz
cd kafka_2.13-3.5.1

启动 Zookeeper

启动 Kafka 代理

步骤 2：为高吞吐量配置 Kafka

修改 config/server.properties 以优化性能

echo "num.network.threads=8" >> config/server.properties
echo "num.io.threads=16" >> config/server.properties
echo "log.dirs=/tmp/kafka-logs" >> config/server.properties
echo "log.retention.hours=48" >> config/server.properties
echo "log.segment.bytes=1073741824" >> config/server.properties
Restart Kafka for changes to take effect:
bin/kafka-server-stop.sh
bin/kafka-server-start.sh config/server.properties

步骤 3：创建 Kafka 主题

创建一个针对高吞吐量进行优化的主题

bin/kafka-topics.sh --create --topic large-scale-events --bootstrap-server localhost:9092 --partitions 10 --replication-factor 3

验证主题创建

bin/kafka-topics.sh --describe --topic large-scale-events --bootstrap-server localhost:9092

输出

步骤 4：运行 Kafka 生产者和消费者

Kafka 生产者（Python）

from kafka import KafkaProducer
import json

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

event = {'event_id': 1, 'type': 'click', 'user': 'Alice'}
producer.send('large-scale-events', value=event)
producer.flush()
print("Sent Event:", event)

输出

Kafka 消费者

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'large-scale-events',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    value_deserializer=lambda v: json.loads(v.decode('utf-8'))
)

for message in consumer:
    print("Received Event:", message.value)

输出

步骤 5：为大规模数据处理进行性能调优

Modify server.properties for high throughput:
echo "num.partitions=10" >> config/server.properties
echo "message.max.bytes=10485760" >> config/server.properties
echo "replica.fetch.max.bytes=10485760" >> config/server.properties
Restart Kafka for changes to apply:
bin/kafka-server-stop.sh
bin/kafka-server-start.sh config/server.properties

为高吞吐量配置 Cosmos DB

设置 Azure Cosmos DB 帐户

使用 Azure CLI 创建 Cosmos DB 帐户

az cosmosdb create --name myCosmosDB --resource-group myResourceGroup --kind GlobalDocumentDB

输出

配置吞吐量设置

启用自动缩放吞吐量

az cosmosdb sql database create --account-name myCosmosDB --resource-group myResourceGroup --name myDatabase --throughput-type autoscale

手动设置预配吞吐量

az cosmosdb sql database throughput update --account-name myCosmosDB --resource-group myResourceGroup --name myDatabase --throughput 10000

输出

创建数据库和容器

创建数据库

az cosmosdb sql database create --account-name myCosmosDB --resource-group myResourceGroup --name myDatabase

创建带分区键的容器

az cosmosdb sql container create --account-name myCosmosDB --resource-group myResourceGroup --database-name myDatabase --name EventsContainer --partition-key-path "/eventType" --throughput 4000

输出

将数据写入 Cosmos DB

Python 代码插入数据

import azure.cosmos.cosmos_client as cosmos_client
import json
ENDPOINT = "https://myCosmosDB.documents.azure.com:443/"
KEY = "your-primary-key"
DATABASE_ID = "myDatabase"
CONTAINER_ID = "EventsContainer"

client = cosmos_client.CosmosClient(ENDPOINT, {'masterKey': KEY})
container = database.get_container_client(CONTAINER_ID)

event = {"id": "1", "eventType": "click", "user": "Alice", "timestamp": "2025-01-29T12:00:00Z"}
container.create_item(event)
print("Inserted Event:", json.dumps(event, indent=2))

输出

优化 Cosmos DB 以实现大规模摄取

启用索引策略以加快查询速度

az cosmosdb sql container update --account-name myCosmosDB --resource-group myResourceGroup --database-name myDatabase --name EventsContainer --indexing-policy "{\"automatic\":true,\"includedPaths\":[{\"path\":\"/*\"}]}"

启用 TTL（生存时间）以实现自动数据过期

az cosmosdb sql container update --account-name myCosmosDB --resource-group myResourceGroup --database-name myDatabase --name EventsContainer --ttl 86400

输出

开发用于大规模事件的 Kafka 生产者

编写高性能 Kafka 生产者

Kafka 生产者负责将数据发送到主题。它连接到代理并实时推送消息。

基本 Kafka 生产者

from kafka import KafkaProducer
import json

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

event = {"event_id": 1, "eventType": "click", "user": "Alice"}
producer.send('large-scale-events', value=event)
producer.flush()
print("Sent Event:", event)

输出

此示例创建一个简单的生产者，它连接到 Kafka 并发送 JSON 格式的事件。

配置生产者优化

为了有效处理大规模事件数据，必须配置诸如批处理、压缩和重试等优化。

优化的 Kafka Producer 配置

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    acks='all',  # Ensures message durability
    compression_type='gzip',  # Compress messages for efficiency
    linger_ms=10,  # Batch messages to improve throughput
    batch_size=32768,  # Increase batch size for high throughput
    max_in_flight_requests_per_connection=5,  # Optimize request handling
    retries=3,
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

优化说明

acks='all'：确保所有副本确认消息后才返回确认。
压缩 (gzip)：减小消息大小，提高吞吐量。
批处理 (linger_ms=10)：等待 10 毫秒收集消息后再发送，提高效率。
重试 (retries=3)：最多重试三次失败的消息。

高效发送大量事件

import time
events = [
    {"event_id": i, "eventType": "play", "user": "User" + str(i)}
    for i in range(1000)
]

start_time = time.time()
for event in events:
    producer.send('large-scale-events', value=event)
producer.flush()
end_time = time.time()

print(f"Sent {len(events)} events in {end_time - start_time:.2f} seconds")

输出

批处理消息通过减少发送到 Kafka 的请求数量来提高性能。

监控和错误处理

Kafka 生产者必须处理诸如代理故障、网络问题或超时等错误。

处理消息发送失败

def on_success(metadata):
    print(f"Message sent to {metadata.topic} on partition {metadata.partition}")

def on_error(excp):
    print(f"Error: {excp}")

for event in events:
    producer.send('large-scale-events', value=event).add_callback(on_success).add_errback(on_error)
producer.flush()

输出（成功和失败的消息传递示例）

关键错误处理策略

回调 (add_callback, add_errback)：提供成功或失败通知。
重试 (retries=3)：自动重试失败的消息。
日志记录：记录失败日志以便于调试和监控。

构建 Kafka 消费者以在 Cosmos DB 中存储事件

编写高效的 Kafka 消费者

Kafka 消费者监听一个主题并读取消息进行处理。

Python 中的基本 Kafka 消费者

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'large-scale-events',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='event-consumer-group',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

for message in consumer:
    print("Received Event:", message.value)

输出

将消费者连接到 Cosmos DB

设置 Cosmos DB 连接

from azure.cosmos import CosmosClient

COSMOS_DB_URL = "<your_cosmos_db_url>"
COSMOS_DB_KEY = "<your_cosmos_db_key>"
DATABASE_NAME = "EventDB"
CONTAINER_NAME = "Events"

client = CosmosClient(COSMOS_DB_URL, COSMOS_DB_KEY)
database = client.create_database_if_not_exists(DATABASE_NAME)
container = database.create_container_if_not_exists(id=CONTAINER_NAME, partition_key="/eventType")

将 Kafka 事件存储到 Cosmos DB

def store_event(event):
    container.upsert_item(event)
    print("Stored Event:", event)

for message in consumer:
    store_event(message.value)

输出

高效处理大规模事件数据

批处理以实现高吞吐量

batch = []
BATCH_SIZE = 100

def store_batch():
    if batch:
        container.upsert_items(batch)
        print(f"Stored {len(batch)} events in Cosmos DB")
        batch.clear()

for message in consumer:
    batch.append(message.value)
    if len(batch) >= BATCH_SIZE:
        store_batch()

输出

监控和错误处理

处理消息处理失败

import logging

logging.basicConfig(level=logging.ERROR)

def store_event_with_error_handling(event):
    try:
        container.upsert_item(event)
    except Exception as e:
        logging.error(f"Failed to store event: {event}, Error: {e}")

for message in consumer:
    store_event_with_error_handling(message.value)

输出（示例错误处理）

为大规模数据处理优化性能

分区策略：在 Cosmos DB 中使用高基数的分区键。
批处理：使用批量摄取而不是逐个插入。
压缩：为 Kafka 消息启用 Gzip 压缩。
扩展 Kafka 消费者：在消费者组中部署多个消费者。
Cosmos DB 索引：定义索引策略以加快查询速度。

下一主题Kafka 主题和 Cosmos DB 容器的分区策略

使用 Kafka 在 Cosmos DB 中处理大规模优化数据

引言

大规模事件处理的重要性

为什么将 Kafka 与 Cosmos DB 结合使用？

1. 可伸缩性和弹性

2. 高可用性和可靠性

3. 实时处理和分析

4. 成本优化

5. 多模型和灵活的模式支持

架构概述

为大规模事件处理设置 Kafka

概述

步骤 1：安装和启动 Kafka

步骤 2：为高吞吐量配置 Kafka

步骤 3：创建 Kafka 主题

步骤 4：运行 Kafka 生产者和消费者

步骤 5：为大规模数据处理进行性能调优

为高吞吐量配置 Cosmos DB

设置 Azure Cosmos DB 帐户

配置吞吐量设置

创建数据库和容器

将数据写入 Cosmos DB

优化 Cosmos DB 以实现大规模摄取

开发用于大规模事件的 Kafka 生产者

编写高性能 Kafka 生产者

配置生产者优化

监控和错误处理

构建 Kafka 消费者以在 Cosmos DB 中存储事件

编写高效的 Kafka 消费者

将消费者连接到 Cosmos DB

监控和错误处理

为大规模数据处理优化性能

联系信息

关注我们

教程

面试题

在线编译器

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 与 Cosmos DB 结合使用？

1. 可伸缩性和弹性

2. 高可用性和可靠性

3. 实时处理和分析

4. 成本优化

5. 多模型和灵活的模式支持

架构概述

为大规模事件处理设置 Kafka

概述

步骤 1：安装和启动 Kafka

步骤 2：为高吞吐量配置 Kafka

步骤 3：创建 Kafka 主题

步骤 4：运行 Kafka 生产者和消费者

步骤 5：为大规模数据处理进行性能调优

为高吞吐量配置 Cosmos DB

设置 Azure Cosmos DB 帐户

配置吞吐量设置

创建数据库和容器

将数据写入 Cosmos DB

优化 Cosmos DB 以实现大规模摄取

开发用于大规模事件的 Kafka 生产者

编写高性能 Kafka 生产者

配置生产者优化

监控和错误处理

构建 Kafka 消费者以在 Cosmos DB 中存储事件

编写高效的 Kafka 消费者

将消费者连接到 Cosmos DB

监控和错误处理

为大规模数据处理优化性能

相关帖子

节能 Kafka 集群

Kafka 消费者重新平衡

使用 Kafka Connect 演进旧系统

Kafka 在增强现实 (AR) 应用程序中的作用

基于 Kafka 架构中的 Cosmos DB 一致性级别

Kafka 到 BigQuery ETL 流程（加载数据）

Kafka 消息压缩

Kafka Streams DSL

Apache Kafka BigQuery 集成

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

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器