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

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

引言

数据更改捕获 (CDC) 是一种用于检测和捕获对数据库所做的更改的技术，以便这些更改可以在其他地方进行处理，例如在数据管道或分析系统中。Apache Kafka 是一个分布式事件流平台，是实现 CDC 管道的流行工具，它提供了可伸缩性、容错性和高吞吐量。Azure Cosmos DB 是一个全球分布式 NoSQL 数据库，由于其可伸缩性和低延迟操作，通常是此类 CDC 管道的目标。

架构概述

该架构包含以下组件

CDC 工具： Debezium 等工具可以从源数据库捕获更改，并将它们流式传输到 Kafka。
Kafka： 作为消息传递层来处理 CDC 事件。
Cosmos DB： 捕获到的更改将被存储的目标数据库。
Kafka 消费者： 从 Kafka 读取 CDC 事件并将其写入 Cosmos DB。

架构流程

为了实现下面的架构流程，我们提供了一个简化的 Python 脚本来模拟以下步骤

模拟源数据库中的更改： 我们使用模拟数据更改，而不是实际的数据库。
将更改发布到 Kafka： 使用 Kafka producer 将 CDC 事件发送到主题。
从 Kafka 消费更改并写入 Cosmos DB： 使用 Kafka consumer 处理这些事件并将它们存储在 Cosmos DB 中。

这是完整的程序

组合的 CDC 程序

from kafka import KafkaProducer, KafkaConsumer
from azure.cosmos import CosmosClient, exceptions
import json
import time
KAFKA_TOPIC = "cdc_topic"
KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
COSMOS_ENDPOINT = "<COSMOS_DB_ENDPOINT>"
COSMOS_KEY = "<COSMOS_DB_KEY>"
DATABASE_NAME = "cdc_demo"
CONTAINER_NAME = "employees"
producer = KafkaProducer(
    bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)
consumer = KafkaConsumer(
    KAFKA_TOPIC,
    bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
    value_deserializer=lambda m: json.loads(m.decode("utf-8")),
    group_id="cdc_group"
)
cosmos_client = CosmosClient(COSMOS_ENDPOINT, COSMOS_KEY)
database = cosmos_client.get_database_client(DATABASE_NAME)
container = database.get_container_client(CONTAINER_NAME)

# Step 1: Simulate Source Database Changes
def simulate_database_changes():
    print("Simulating database changes...")
    changes = [
        {"id": 1, "name": "Alice", "position": "Engineer", "salary": 70000},
        {"id": 2, "name": "Bob", "position": "Manager", "salary": 90000},
        {"id": 1, "name": "Alice", "position": "Senior Engineer", "salary": 80000},  
# Update
    ]
    for change in changes:
        print(f"Publishing change: {change}")
        producer.send(KAFKA_TOPIC, value=change)
        time.sleep(1)  # Simulate time delay between changes
def process_events():
    print("Starting Kafka consumer...")
    for message in consumer:
        cdc_event = message.value
        print(f"Received event: {cdc_event}")
        try:
            container.upsert_item(cdc_event)
            print(f"Record upserted in Cosmos DB: {cdc_event}")
        except exceptions.CosmosHttpResponseError as e:
            print(f"Failed to upsert record: {e}")
# Main Execution
if __name__ == "__main__":
    # Run both steps
    from threading import Thread

    producer_thread = Thread(target=simulate_database_changes)
    consumer_thread = Thread(target=process_events)

    producer_thread.start()
    consumer_thread.start()

    producer_thread.join()
    consumer_thread.join()   

示例输出

模拟数据库更改

Change Data Capture (CDC) from Kafka to Cosmos DB

Kafka 消费者和 Cosmos DB 写入

程序说明

模拟数据库更改
- simulate_database_changes 函数将模拟的 CDC 事件生成为 JSON 对象，并使用 KafkaProducer 将它们发送到 Kafka 主题。
发布到 Kafka
- 每个更改都会被序列化为 JSON 并发送到 cdc_topic Kafka 主题。
消费事件
- KafkaConsumer 监听 cdc_topic 并处理每个事件。
写入 Cosmos DB
- container.upsert_item 方法将事件写入 Cosmos DB，如果记录存在则更新，如果不存在则插入。

要点

单个程序： 此脚本整合了整个 CDC 管道流程，以简化操作。
实时处理： 该管道在事件发生时处理它们，并近实时地更新 Cosmos DB。
可伸缩架构： 您可以通过添加更多 Kafka 分区、消费者或 Cosmos DB 容器来按需扩展此架构。

设置 Kafka

步骤 1：安装 Kafka

下载并安装 Kafka。请按照官方文档进行设置。以下是本地 Kafka 安装的快速入门指南

# Download Kafka
wget https://downloads.apache.org/kafka/3.5.1/kafka_2.13-3.5.1.tgz

# Extract Kafka
tar -xzf kafka_2.13-3.5.1.tgz
cd kafka_2.13-3.5.1
bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties   

步骤 2：创建 Kafka 主题

为 CDC 事件创建一个主题

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

源数据库设置

步骤 1：设置示例数据库

在本指南中，我们使用 MySQL 作为源数据库

CREATE DATABASE cdc_demo;
USE cdc_demo;

CREATE TABLE employees (
  id INT AUTO_INCREMENT PRIMARY KEY,
  name VARCHAR(100),
  position VARCHAR(50),
  salary DECIMAL(10, 2),
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

INSERT INTO employees (name, position, salary) VALUES
('Alice', 'Engineer', 70000),
('Bob', 'Manager', 90000);   

步骤 2：配置 Debezium 以进行 CDC

Debezium 是一个 CDC 工具，与 Kafka 集成良好。按照以下步骤设置 Debezium

将 Debezium MySQL 连接器插件添加到您的 Kafka Connect 安装中。
配置 Debezium MySQL 连接器

{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "localhost",
    "database.port": "3306",
    "database.user": "root",
    "database.password": "password",
    "database.server.id": "184054",
    "database.server.name": "mysql_server",
    "database.include.list": "cdc_demo",
    "table.include.list": "cdc_demo.employees",
    "database.history.kafka.bootstrap.servers": "localhost:9092",
    "database.history.kafka.topic": "schema-changes.cdc_demo"
  }
}   

在 Kafka Connect 中部署此配置，以开始将更改从 MySQL 流式传输到 Kafka。

Azure Cosmos DB 设置

步骤 1：创建 Cosmos DB 实例

登录到 Azure 门户。
创建一个新的 Cosmos DB 帐户。选择 API 为 Core (SQL)。
创建数据库和容器
- 数据库名称：cdc_demo
- 容器名称：employees
- 分区键：/id

步骤 2：记下连接详细信息

从 Azure 门户检索 Cosmos DB 连接字符串和密钥。

代码实现

以下是一个基于 Python 的 Kafka 消费者，它将 CDC 事件提取到 Cosmos DB 中

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

# Cosmos DB configuration
COSMOS_ENDPOINT = "<COSMOS_DB_ENDPOINT>"
COSMOS_KEY = "<COSMOS_DB_KEY>"
DATABASE_NAME = "cdc_demo"
CONTAINER_NAME = "employees"

# Kafka configuration
KAFKA_TOPIC = "cdc_topic"
KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"

# Initialize Cosmos DB client
cosmos_client = CosmosClient(COSMOS_ENDPOINT, COSMOS_KEY)
database = cosmos_client.get_database_client(DATABASE_NAME)
container = database.get_container_client(CONTAINER_NAME)

# Initialize Kafka consumer
consumer = KafkaConsumer(
    KAFKA_TOPIC,
    bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

print("Starting Kafka consumer...")

for message in consumer:
    cdc_event = message.value
    print(f"Received event: {cdc_event}")

    # Insert or update record in Cosmos DB
    try:
        container.upsert_item(cdc_event)
        print(f"Record upserted: {cdc_event}")
    except exceptions.CosmosHttpResponseError as e:
        print(f"Failed to upsert record: {e}")   

说明

KafkaConsumer： 从 Kafka 主题读取消息。
CosmosClient： 与 Cosmos DB 交互以插入或更新记录。
Upsert Item： 确保记录根据其 id 被插入或更新。

端到端流程

步骤 1：测试 CDC 管道

1. 在 MySQL 中插入一条新记录

 
INSERT INTO employees (name, position, salary) VALUES ('Charlie', 'Analyst', 60000);   

2. 观察 Kafka 消费者日志

3. 收到的事件：{"id": 3, "name": "Charlie", "position": "Analyst", "salary": 60000}

 
Record upserted: {"id": 3, "name": "Charlie", "position": "Analyst", "salary": 60000}   

4. 使用 Azure 门户或 SDK 在 Cosmos DB 中验证记录。

步骤 2：更新记录

在 MySQL 中更新一条记录

Kafka 消费者日志

Received event: {"id": 1, "name": "Alice", "position": "Engineer", "salary": 75000}
Record upserted: {"id": 1, "name": "Alice", "position": "Engineer", "salary": 75000}   

最佳实践

分区： 在 Cosmos DB 中使用逻辑分区键来优化查询性能。
错误处理： 为瞬态错误实现重试机制。
监控： 使用 Prometheus 和 Grafana 等工具监控管道。
模式演进： 为源数据库中的模式更改做好计划。

下一主题Cosmos-db-as-a-sink-for-apache-kafka-topics

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

引言

架构概述

架构流程

设置 Kafka

步骤 1：安装 Kafka

步骤 2：创建 Kafka 主题

源数据库设置

步骤 1：设置示例数据库

步骤 2：配置 Debezium 以进行 CDC

Azure Cosmos DB 设置

步骤 1：创建 Cosmos DB 实例

步骤 2：记下连接详细信息

端到端流程

步骤 1：测试 CDC 管道

步骤 2：更新记录

最佳实践

联系信息

关注我们

教程

面试题

在线编译器

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 的变更数据捕获 (CDC)

引言

架构概述

架构流程

设置 Kafka

步骤 1：安装 Kafka

步骤 2：创建 Kafka 主题

源数据库设置

步骤 1：设置示例数据库

步骤 2：配置 Debezium 以进行 CDC

Azure Cosmos DB 设置

步骤 1：创建 Cosmos DB 实例

步骤 2：记下连接详细信息

端到端流程

步骤 1：测试 CDC 管道

步骤 2：更新记录

最佳实践

相关帖子

Kafka CLI 和工具

使用 Kafka 的流优先架构

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

Kafka 作为智能城市的中枢神经系统

Kafka 代理 - 详细信息

Kafka Streams 的时间旅行查询

Kafka 在网络安全事件管理中的作用

Kafka 中的日志压缩

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

模式演进

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器