Cosmos DB 作为 Apache Kafka 主题的接收器

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

将 Cosmos DB 用作 Apache Kafka 主题的接收器是一个强大的组合，它允许将实时流数据摄取到像 Azure Cosmos DB 这样全球分布式、多模型数据库中。这使得企业能够高效地处理、分析和可视化数据。

1. Cosmos DB 和 Kafka 简介

Cosmos DB 概述

Azure Cosmos DB 是一个完全托管的 NoSQL 数据库服务，专为现代应用程序开发而设计。主要功能包括：

多模型支持：支持文档、键值、图和列族数据模型。
全球分布式 允许跨多个区域进行复制。
低延迟：保证个位数毫秒级的响应时间。

Apache Kafka 概述

Apache Kafka 是一个分布式事件流平台，用于构建实时数据管道和流应用程序。主要组件包括：

生产者：将数据发布到 Kafka 主题。
主题：对数据流进行分类和存储。
消费者：从主题读取和处理数据。

为什么使用 Cosmos DB 作为接收器？

实时数据处理：将流数据存储在 Cosmos DB 中，用于分析和可视化。
可扩展性：处理高速度的数据流。
无缝集成：利用 Cosmos DB Kafka 连接器实现平稳摄取。

2. 架构概述

将 Cosmos DB 用作接收器的典型架构包括以下组件：

Kafka 生产者

生产者将实时事件发送到 Kafka 主题。生产者通常使用 Java 或 Python 等编程语言中的 Kafka 库来实现。

Kafka 主题

主题作为流数据的存储单元。生产者写入主题，消费者（如 Cosmos DB Sink Connector）从中读取。

Kafka Connect

Kafka Connect 简化了 Kafka 与外部系统的集成。Cosmos DB Sink Connector 是 Kafka Connect 的一个插件。

Cosmos DB Sink Connector

Cosmos DB Sink Connector 从 Kafka 主题读取数据并将其写入 Cosmos DB 容器。它负责将 Kafka 主题模式映射到 Cosmos DB 文档结构。

3. 环境设置

前提条件

Azure 订阅：可以访问 Azure Portal 来设置 Cosmos DB。
Apache Kafka：已安装并在本地或云环境中运行。
Kafka Connect：作为 Kafka 生态系统的一部分已安装。
Cosmos DB Kafka Connector：从 Confluent Hub 下载。
Java Development Kit (JDK)：用于生产者开发。
Maven 或 Gradle：用于依赖管理。

设置 Apache Kafka

1. 安装 Kafka

2. tar -xvf kafka_2.13-3.5.1.tgz

3. 启动 Kafka

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

4. 创建一个主题

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

配置 Azure Cosmos DB

创建 Cosmos DB 帐户
- 导航到 Azure 门户。
- 选择 **Azure Cosmos DB** 并创建一个新帐户，使用 **Core (SQL)**（SQL）
创建数据库和容器
- 创建一个数据库（例如，kafka_db）。
- 添加一个容器（例如，kafka_data），并指定一个分区键（例如，/id）。
获取连接详细信息
- 从 Cosmos DB 帐户复制 **Primary Key** 和 **Connection String**。

4. 开发 Kafka 生产者

Apache Kafka 是一个强大的消息和流平台，它促进了生产者和消费者之间无缝的数据交换。当 Cosmos DB 用作接收器时，Kafka 成为摄取、处理并将数据传输到 Azure Cosmos DB 进行存储和分析的通道。

实现 Kafka 生产者以用于 Cosmos DB 的步骤

设置 Kafka 和 Cosmos DB 环境
- 安装 Apache Kafka，并确保 Broker 和 Zookeeper 正在运行。
- 设置 Cosmos DB，并使用 Kafka Connect 框架将其配置为充当 Kafka 接收器。
将 Kafka Connect 配置为使用 Cosmos DB Sink
- 使用 Cosmos DB sink 连接器。
- 使用端点 URL、访问密钥和数据库名称等必要参数配置连接器。
在 Java 中开发 Kafka 生产者
- 创建一个 Java 应用程序，该应用程序生成样本数据并将其发送到 Kafka 主题。

前提条件

已安装 Java Development Kit (JDK)（8 或更高版本）。
已安装 Apache Kafka，并在本地或服务器上运行。
拥有创建容器和数据库权限的 Azure Cosmos DB 帐户。
已安装带有 Cosmos DB Sink Connector 的 Kafka Connect。

Code Example

1. Maven 项目结构

确保您的项目使用 Maven 来管理依赖项。以下是一个示例 pom.xml：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>KafkaProducerExample</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.5.0</version>
        </dependency>
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.9.0</version>
        </dependency>
    </dependencies>
</project>   

2. Kafka 生产者实现

创建一个用于 Kafka 生产者的 Java 类。

package com.example.kafkaproducer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

// Main Kafka Producer class
public class SampleKafkaProducer {

    public static void main(String[] args) {
        // Kafka configuration
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092"); // Kafka broker
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "all");
        Producer<String, String> producer = new KafkaProducer<>(props);
        String topic = "cosmosdb_topic";
        try {
            for (int i = 1; i <= 10; i++) {
                String key = "key-" + i;
                String value = generateSampleData(i);
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

                // Synchronous send
                System.out.printf("Sent record(key=%s value=%s) meta(partition=%d, offset=%d)%n",
                        key, value, metadata.partition(), metadata.offset());
            }
        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }

    // Generate sample data in JSON format
                RecordMetadata metadata = producer.send(record).get();
    private static String generateSampleData(int id) {
        return String.format("{\"id\": \"%d\", \"name\": \"Sample Item %d\", \"category\": \"Category %d\", \"timestamp\": \"%d\"}",
                id, id, id % 3, System.currentTimeMillis());
    }
}

3. 使用 Cosmos DB Sink 配置 Kafka Connect

1. 安装 Cosmos DB Sink Connector。

下载 Cosmos DB Kafka Connector JAR 文件，并将其放置在 Kafka 的 libs 目录中。

2. 使用属性文件（cosmosdb-sink.properties）配置 sink 连接器。

name=cosmosdb-sink-connector
connector.class=com.azure.cosmos.kafka.connect.sink.CosmosDBSinkConnector
tasks.max=1
topics=cosmosdb_topic
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter.schemas.enable=false

# Cosmos DB-specific configuration
connect.cosmos.connection.endpoint=<YOUR_COSMOS_DB_ENDPOINT>
connect.cosmos.connection.key=<YOUR_COSMOS_DB_ACCESS_KEY>
connect.cosmos.database.name=KafkaSinkDemoDB
connect.cosmos.container.name=KafkaSinkDemoContainer   

3. 启动 Kafka Connect worker 并加载连接器配置。

 
bin/connect-standalone.sh config/connect-standalone.properties config/cosmosdb-sink.properties   

4. 运行示例

1. 启动 Kafka 组件

启动 Zookeeper 和 Kafka broker。
使用 kafka-topics.sh 验证 Kafka 主题。

bin/kafka-topics.sh --create --topic cosmosdb_topic --bootstrap-server localhost:9092   

2. 运行生产者

编译并运行生产者代码。
检查 Kafka 日志以验证消息是否已生成。

3. 验证 Cosmos DB

检查 Cosmos DB 容器以查看已摄取的数据。

示例输出

生产者控制台输出

Cosmos DB as a Sink for Apache Kafka Topics

Cosmos DB 数据

5. 配置 Cosmos DB Sink Connector

安装和设置

可以使用 Confluent Hub CLI 安装 Cosmos DB Sink Connector。

安装后，通过启动 Kafka Connect worker 来确保其正在运行。

配置参数

创建一个 Cosmos DB Sink Connector 的配置文件（例如，cosmos_sink_connector.json），其中包含以下详细信息：

{
    "name": "cosmosdb-sink-connector",
    "config": {
        "connector.class": "com.azure.cosmos.kafka.connect.sink.CosmosDBSinkConnector",
        "tasks.max": "1",
        "topics": "cosmos_sink_topic",
        "cosmos.connection.endpoint": "<YOUR_COSMOS_DB_URI>",
        "cosmos.connection.key": "<YOUR_COSMOS_DB_PRIMARY_KEY>",
        "cosmos.database.name": "MusicDB",
        "cosmos.container.name": "Plays",
        "cosmos.key.converter": "org.apache.kafka.connect.storage.StringConverter",
         value.converter=org.apache.kafka.connect.json.JsonConverter,
        "cosmos.value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "cosmos.value.converter.schemas.enable": "false",
        "cosmos.max.batch.size": "100",
        "cosmos.batch.flush.timeout.ms": "5000"
    }
}   

关键参数说明

connector.class：指定要使用的连接器类。
tasks.max：连接器的任务数量。
topics：连接器将从中读取的 Kafka 主题。
cosmos.connection.endpoint：Cosmos DB 帐户的 URI。
cosmos.connection.key：Cosmos DB 帐户的主密钥。
cosmos.database.name：目标 Cosmos DB 数据库。
cosmos.container.name：目标 Cosmos DB 数据库中的容器。
cosmos.key.converter 和 cosmos.value.converter：键和值的格式转换器。
cosmos.max.batch.size：每个批次的最大记录数。
cosmos.batch.flush.timeout.ms：刷新批次之前的超时时间（以毫秒为单位）。

下一主题Kafka-connect-bigquery-sink

Cosmos DB 作为 Apache Kafka 主题的接收器

1. Cosmos DB 和 Kafka 简介

Cosmos DB 概述

Apache Kafka 概述

为什么使用 Cosmos DB 作为接收器？

2. 架构概述

Kafka 生产者

Kafka 主题

Kafka Connect

Cosmos DB Sink Connector

3. 环境设置

前提条件

设置 Apache Kafka

配置 Azure Cosmos DB

4. 开发 Kafka 生产者

实现 Kafka 生产者以用于 Cosmos DB 的步骤

前提条件

Code Example

5. 配置 Cosmos DB Sink Connector

安装和设置

配置参数

联系信息

关注我们

教程

面试题

在线编译器

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

杂项

Cosmos DB 作为 Apache Kafka 主题的接收器

1. Cosmos DB 和 Kafka 简介

Cosmos DB 概述

Apache Kafka 概述

为什么使用 Cosmos DB 作为接收器？

2. 架构概述

Kafka 生产者

Kafka 主题

Kafka Connect

Cosmos DB Sink Connector

3. 环境设置

前提条件

设置 Apache Kafka

配置 Azure Cosmos DB

4. 开发 Kafka 生产者

实现 Kafka 生产者以用于 Cosmos DB 的步骤

前提条件

Code Example

5. 配置 Cosmos DB Sink Connector

安装和设置

配置参数

相关帖子

Kafka 保留策略

Kafka 用于跨数据中心复制的挑战

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

Kafka 偏移量管理

最佳 Apache Kafka 书籍

Kafka 在法规遵从性自动化中的应用

无服务器 Kafka 应用程序的优势

消费者在 Kafka 生态系统中的作用

Apache Kafka vs RabbitMQ

Kafka 中的实时欺诈检测管道

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器