Kafka Connect for Azure Cosmos DB - 源连接器 v2

适用范围： NoSQL

Kafka Connect for Azure Cosmos DB 是用于从/向 Azure Cosmos DB 读取/写入数据的连接器。 Azure Cosmos DB 源连接器提供从 Azure Cosmos DB 更改源读取数据并将这些数据发布到 Kafka 主题的功能。

• 请从 Confluent 平台设置开始，因为完成此过程可以提供一个可用的完整环境。 如果你不想要使用 Confluent 平台，则需要自行安装并配置 Zookeeper、Apache Kafka 和 Kafka Connect。 还需要手动安装并配置 Azure Cosmos DB 连接器。
• 按照容器设置指南创建一个 Azure Cosmos DB 帐户
• Bash shell
• 下载 Java 11+
• 下载 Maven

如果使用的是建议的 Confluent 平台设置，则 Azure Cosmos DB 源连接器已包含在安装中，因此可以跳过此步骤。

否则，可以使用最新版本中的 JAR 文件，并手动安装连接器。 有关详细信息，请参阅这些说明。 还可以从源代码打包新的 JAR 文件：

# clone the azure-sdk-for-java repo if you haven't done so already
git clone https://github.com/Azure/azure-sdk-for-java.git
cd sdk/cosmos

mvn -e -DskipTests -Dgpg.skip -Dmaven.javadoc.skip=true -Dcodesnippet.skip=true -Dspotbugs.skip=true -Dcheckstyle.skip=true -Drevapi.skip=true -pl ,azure-cosmos,azure-cosmos-tests -am clean install
mvn -e -DskipTests -Dgpg.skip -Dmaven.javadoc.skip=true -Dcodesnippet.skip=true -Dspotbugs.skip=true -Dcheckstyle.skip=true -Drevapi.skip=true -pl ,azure-cosmos-kafka-connect clean install

# include the following JAR file in Kafka Connect installation
ls target/azure-cosmos-kafka-connect-*.jar

使用 Confluent 控制中心创建 Kafka 主题。 对于此方案，我们将创建一个名为“apparels”的 Kafka 主题，并将非架构嵌入式 JSON 数据写入该主题。 若要在控制中心内创建主题，请参阅创建 Kafka 主题文档。

若要在 Kafka Connect 中创建 Azure Cosmos DB 源连接器，请使用以下 JSON 配置。确保替换 azure.cosmos.account.endpoint 和 azure.cosmos.account.key 属性的占位符值（在学习先决条件部分所述的 Azure Cosmos DB 设置指南时已保存）。

{
  "name": "cosmosdb-source-connector-v2",
  "config": {
    "connector.class": "com.azure.cosmos.kafka.connect.CosmosSourceConnector",
    "tasks.max": "5",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "azure.cosmos.account.endpoint":"{endpoint}",
    "azure.cosmos.account.key":"{masterKey}",
    "azure.cosmos.application.name": "{applicationName}",
    "azure.cosmos.source.database.name":"{database}",
    "azure.cosmos.source.containers.includedList":"{container}",
    "azure.cosmos.source.changeFeed.maxItemCountHint":"500",
    "azure.cosmos.source.containers.topicMap":"{topic}#{container}",
    "azure.cosmos.source.metadata.storage.type":"Cosmos",
    "azure.cosmos.source.metadata.storage.name":"{metadataContainerName}"
  }
}

有关上述每个配置属性的详细信息，请参阅源属性部分。 填写所有值后，将 JSON 文件保存在本地的某个位置。 可以使用此文件通过 REST API 创建连接器。

创建连接器的简单方法是使用 Confluent 控制中心门户。 请按照 Confluent 设置指南从控制中心创建连接器。 设置时，请不要使用 DatagenConnector 选项，而应改用 CosmosDBSourceConnector 磁贴。 配置源连接器时，请填写在 JSON 文件中填充的值。

或者，也可以在“连接器”页中，使用“上传连接器配置文件”选项上传在前一部分生成的 JSON 文件。

使用 Kafka“连接”REST API 创建源连接器

# Curl to Kafka connect service
curl -H "Content-Type: application/json" -X POST -d @<path-to-JSON-config-file> http://localhost:8083/connectors

1. 登录到 Azure 门户并导航到你的 Azure Cosmos DB 帐户。

2. 打开“数据资源管理器”选项卡并选择“数据库”

3. 打开前面创建的“kafkaconnect”数据库和“kafka”容器。

4. 若要创建新的 JSON 文档，请在“API for NoSQL”窗格中展开“kafka”容器，选择“项”，然后在工具栏中选择“新建项”。

5. 现在，将文档添加到采用以下结构的容器。 将以下示例 JSON 块粘贴到“项”选项卡中，并覆盖当前内容：

   
   {
     "id": "2",
     "productId": "33218897",
     "category": "Women's Outerwear",
     "manufacturer": "Contoso",
     "description": "Black wool pea-coat",
     "price": "49.99",
     "shipping": {
       "weight": 2,
       "dimensions": {
         "width": 8,
         "height": 11,
         "depth": 3
       }
     }
   }
   
   

6. 选择“保存”。

7. 查看左侧菜单中的“项”，确认文档已保存。

1. 通过 http://localhost:9000 打开 Kafka 主题 UI。
2. 选择创建的 Kafka“apparels”主题。
3. 检查前面已插入到 Azure Cosmos DB 的文档是否显示在该 Kafka 主题中。

若要在 Confluent 控制中心删除连接器，请导航到创建的源连接器，然后选择“删除”图标。

或者，使用连接器的 REST API：

# Curl to Kafka connect service
curl -X DELETE http://localhost:8083/connectors/cosmosdb-source-connector

若要使用 Azure CLI 删除创建的 Azure Cosmos DB 服务及其资源组，请参阅这些步骤。

以下设置用于配置 Kafka 源连接器。 这些配置值确定了要使用哪个 Azure Cosmos DB 容器、要从哪些 Kafka 主题写入数据，以及用于序列化数据的格式。 有关使用默认值的示例，请参阅此配置文件。

Azure Cosmos DB 源连接器将 JSON 文档转换为架构，支持以下 JSON 数据类型：

• Kafka Connect for Azure Cosmos DB 接收器连接器