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

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

1. 端到端数据流简介

在现代基于云的架构中，实时数据流对于处理大规模事件驱动数据至关重要。Apache Kafka、Azure Functions 和 Azure Cosmos DB 的结合为实现低延迟、可扩展且高度可用的数据流解决方案提供了强大的支持。

1.1. 实时数据流的需求

传统的批处理方法会引入延迟，不适用于欺诈检测、个性化推荐和物联网遥测分析等实时应用。流式处理架构使企业能够处理和分析生成的数据。

1.2. 流式处理管道的组成部分

典型的实时数据流管道由以下部分组成：

数据生产者

数据生产者从各种来源（如用户活动、IoT 传感器和金融交易）生成实时数据。这些生产者将数据发送到 Kafka 主题。

Python Kafka 生产者代码

from kafka import KafkaProducer
import json
import time

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

producer = create_producer()

data = [
    {"user_id": 1, "event": "play_song", "timestamp": "2025-02-19T12:00:00Z"},
    {"user_id": 2, "event": "pause_song", "timestamp": "2025-02-19T12:01:00Z"},
    {"user_id": 3, "event": "like_song", "timestamp": "2025-02-19T12:02:00Z"}
]

for record in data:
    producer.send('user-events', record)
    time.sleep(1)  # Simulating real-time data flow

producer.flush()
print("Messages sent successfully")

预期输出

End-to-End Data Streaming with Kafka, Azure Functions, and Cosmos DB

消息代理 (Kafka)

Kafka 作为分布式消息代理，负责捕获和传输数据流。Kafka 主题 "user-events" 存储用户活动数据。

Kafka 主题创建

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

启动 Kafka 消费者以验证数据

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

预期输出

流处理层 (Azure Functions)

Azure Functions 在将数据发送到 Cosmos DB 之前，会对其进行处理、转换或过滤。

Azure Function for Kafka Trigger

import json
import logging
from azure.cosmos import CosmosClient

COSMOS_DB_ENDPOINT = "<your-cosmos-db-endpoint>"
COSMOS_DB_KEY = "<your-cosmos-db-key>"
DATABASE_NAME = "user-events-db"
CONTAINER_NAME = "events"

def main(event: str):
    client = CosmosClient(COSMOS_DB_ENDPOINT, COSMOS_DB_KEY)
    database = client.create_database_if_not_exists(DATABASE_NAME)
    container = database.create_container_if_not_exists(CONTAINER_NAME, partition_key="/user_id")
    container.create_item(json.loads(event))
    logging.info("Stored event in Cosmos DB.")

日志中的预期输出

存储 (Cosmos DB)

处理后的事件存储在 Azure Cosmos DB 中，用于分析和查询。

查询 Cosmos DB

query = "SELECT * FROM c WHERE c.user_id = 1"
results = list(container.query_items(query, enable_cross_partition_query=True))
print(results)

预期输出

1.3. Kafka、Azure Functions 和 Cosmos DB 的优势

可扩展性：Kafka 可处理高吞吐量的事件摄取，Azure Functions 自动扩展，Cosmos DB 支持弹性扩展。
容错性：Kafka 确保消息持久性，Azure Functions 内置重试机制，Cosmos DB 提供多区域复制。
事件驱动处理：Azure Functions 异步响应 Kafka 事件。

1.4. 用例

实时分析：跟踪流媒体服务中的用户行为。
IoT 数据处理：监控智能设备和工业传感器。
金融交易：银行应用程序中的欺诈检测。

2. Apache Kafka 概述

Apache Kafka 是一个分布式事件流平台，专为高吞吐量、容错和可扩展的实时数据处理而设计。它包含：

2.1. Kafka 核心组件

生产者：将数据发布到 Kafka 主题。
主题：流式数据的逻辑容器。
代理：管理数据分发的 Kafka 服务器。
消费者：订阅并处理主题中数据的应用程序。
ZooKeeper：管理元数据、领导者选举和配置。

2.2. Kafka 架构

Kafka 遵循分布式架构，其中：

数据写入主题。
主题被划分为分区以实现可扩展性。
生产者将数据写入分区。
消费者根据消费者组从分区读取数据。

2.3. 设置 Kafka

安装 Kafka

wget https://downloads.apache.org/kafka/3.2.0/kafka_2.13-3.2.0.tgz
tar -xvf kafka_2.13-3.2.0.tgz
cd kafka_2.13-3.2.0

启动 Kafka 和 ZooKeeper

bin/zookeeper-server-start.sh config/zookeeper.properties &
bin/kafka-server-start.sh config/server.properties &

创建 Kafka 主题

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

3. Azure Functions 简介

Azure Functions 是一项无服务器计算服务，允许用户在不管理基础设施的情况下运行事件驱动的代码。它支持包括 Kafka 在内的多种触发器，非常适合处理流式数据。

3.1. Azure Functions 执行模型

消耗计划：根据调用次数自动扩展。
高级计划：提供增强的性能和扩展能力。
专用 (App Service) 计划：提供对执行的更多控制。

3.2. 创建 Azure Function

安装 Azure Functions Core Tools

创建新的 Azure Function App

func init kafka-consumer --worker-runtime python
cd kafka-consumer
func new --name KafkaFunction --template "KafkaTrigger" --language python

4. 理解 Azure Cosmos DB

Azure Cosmos DB 是一个全球分布式的 NoSQL 数据库，提供高可用性、可扩展性和多模型支持。它针对事件驱动架构进行了优化，并支持：

4.1. 主要功能

多区域复制：确保低延迟访问。
五种一致性模型：强一致性、边界陈旧性、会话一致性、一致性前缀和最终一致性。
自动索引：提高查询性能。

5. Kafka-Azure Functions-Cosmos DB 管道架构

5.1. 数据流

Kafka 生产者：将消息（例如，用户活动日志）发布到 Kafka 主题。
Kafka 代理：将消息路由到适当的分区。
Azure Functions Kafka Trigger：监听 Kafka 主题并处理数据。
Azure Cosmos DB：存储处理后的数据以供分析和操作使用。

5.2. 数据流实现

Kafka Producer Code (Sending Messages to Kafka)

from kafka import KafkaProducer
import json
import time

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

producer = create_producer()

data = [{"user_id": i, "event": "play_song", "timestamp": "2025-02-19T12:00:00Z"} for i in range(1, 4)]

for record in data:
    producer.send('user-events', record)
    time.sleep(1)

producer.flush()
print("Messages sent successfully")

Azure Function (Processing Kafka Messages and Writing to Cosmos DB)

def main(event: str):
    logging.info(f"Processing event: {event}")
    database = client.get_database_client(DATABASE_NAME)
    container = database.get_container_client(CONTAINER_NAME)
    container.create_item(json.loads(event))
    logging.info("Stored event in Cosmos DB.")

预期的 Cosmos DB 查询结果

6. 实现 Kafka 生产者

6.1. 在 Python 中设置 Kafka 生产者

from kafka import KafkaProducer
import json

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

producer = create_producer()
producer.send('user-events', {'user_id': 1, 'event': 'login', 'timestamp': '2025-02-19T12:34:56Z'})
producer.flush()
print("Message sent successfully")

预期输出

7. 使用 Azure Functions 实现 Kafka 消费者

7.1. Azure Function Kafka Trigger

import json
import logging
from azure.cosmos import CosmosClient

COSMOS_DB_ENDPOINT = "<your-cosmos-db-endpoint>"
COSMOS_DB_KEY = "<your-cosmos-db-key>"
DATABASE_NAME = "user-events-db"
CONTAINER_NAME = "events"

def main(event: str):
    logging.info(f"Received event: {event}")
    client = CosmosClient(COSMOS_DB_ENDPOINT, COSMOS_DB_KEY)
    database = client.create_database_if_not_exists(DATABASE_NAME)
    container = database.create_container_if_not_exists(CONTAINER_NAME, partition_key="/user_id")
    container.create_item(json.loads(event))
    logging.info("Stored event in Cosmos DB.")

预期输出

8. 查询 Cosmos DB 以获取处理后的数据

query = "SELECT * FROM c WHERE c.user_id = 1"
results = list(container.query_items(query, enable_cross_partition_query=True))
print(results)

8.1. 预期输出

下一主题Kafka 到 Cosmos DB 数据流的延迟优化

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

1. 端到端数据流简介

1.1. 实时数据流的需求

1.2. 流式处理管道的组成部分

1.3. Kafka、Azure Functions 和 Cosmos DB 的优势

1.4. 用例

2. Apache Kafka 概述

2.1. Kafka 核心组件

2.2. Kafka 架构

2.3. 设置 Kafka

3. Azure Functions 简介

3.1. Azure Functions 执行模型

3.2. 创建 Azure Function

4. 理解 Azure Cosmos DB

4.1. 主要功能

5. Kafka-Azure Functions-Cosmos DB 管道架构

5.1. 数据流

5.2. 数据流实现

6. 实现 Kafka 生产者

6.1. 在 Python 中设置 Kafka 生产者

7. 使用 Azure Functions 实现 Kafka 消费者

7.1. Azure Function Kafka Trigger

8. 查询 Cosmos DB 以获取处理后的数据

8.1. 预期输出

联系信息

关注我们

教程

面试题

在线编译器

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、Azure 函数和 Cosmos DB 进行端到端数据流

1. 端到端数据流简介

1.1. 实时数据流的需求

1.2. 流式处理管道的组成部分

1.3. Kafka、Azure Functions 和 Cosmos DB 的优势

1.4. 用例

2. Apache Kafka 概述

2.1. Kafka 核心组件

2.2. Kafka 架构

2.3. 设置 Kafka

3. Azure Functions 简介

3.1. Azure Functions 执行模型

3.2. 创建 Azure Function

4. 理解 Azure Cosmos DB

4.1. 主要功能

5. Kafka-Azure Functions-Cosmos DB 管道架构

5.1. 数据流

5.2. 数据流实现

6. 实现 Kafka 生产者

6.1. 在 Python 中设置 Kafka 生产者

7. 使用 Azure Functions 实现 Kafka 消费者

7.1. Azure Function Kafka Trigger

8. 查询 Cosmos DB 以获取处理后的数据

8.1. 预期输出

相关帖子

Kafka 中的 Zookeeper：详细信息

Kafka 超越发布-订阅：高级用法模式

Kafka 与区块链系统集成

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

使用 Kafka 的预测性缓存用例

Kafka 中的事件流版本控制

使用 Kafka Connect 演进旧系统

Kafka 代理 - 详细信息

AI 驱动的 Kafka 代理优化

Kafka Streams vs Spark Streaming

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器