Przenoszenie danych przy użyciu łącznika platformy Kafka

Apache Kafka to platforma typu open source używana do przesyłania strumieniowego zdarzeń w sposób rozproszony. Wiele firm korzysta z platformy Kafka na potrzeby scenariuszy integracji danych o wysokiej wydajności na dużą skalę. Kafka Connect to narzędzie w swoim pakiecie służące do przesyłania strumieniowego danych między platformą Kafka a innymi systemami danych. Co zrozumiałe, może to obejmować usługę Azure Cosmos DB jako źródło danych lub docelowy (ujście) danych.

Ustawienia

Łączniki platformy Kafka Connect dla usługi Azure Cosmos DB są dostępne jako projekt open source w usłudze GitHub w witrynie microsoft/kafka-connect-cosmosdb. Instrukcje dotyczące ręcznego pobierania i instalowania pliku JAR są dostępne w repozytorium.

Konfigurowanie

Należy ustawić cztery właściwości konfiguracji, aby prawidłowo skonfigurować łączność z kontem usługi Azure Cosmos DB for NoSQL.

Tematy do mapowania kontenerów

Każdy kontener powinien być mapowany na temat. Załóżmy na przykład, że chcesz, aby kontener products został zamapowany na temat prodlistener i kontener customers do tematu custlistener. W takim przypadku należy użyć następującego ciągu mapowania CSV: prodlistener#products,custlistener#customers.

Zapisywanie w usłudze Azure Cosmos DB

Napiszmy dane do usługi Azure Cosmos DB, tworząc temat. Na platformie Apache Kafka wszystkie komunikaty są wysyłane za pośrednictwem tematów.

Nowy temat można utworzyć przy użyciu polecenia kafka-topics . Ten przykład spowoduje utworzenie nowego tematu o nazwie prodlistener.

kafka-topics --create \
    --zookeeper localhost:2181 \
    --topic prodlistener \
    --replication-factor 1 \
    --partitions 1

Następujące polecenie uruchomi producenta, aby można było zapisać trzy rekordy w temacie prodlistener.

kafka-console-producer \
    --broker-list localhost:9092 \
    --topic prodlistener

W konsoli programu można następnie wprowadzić te trzy rekordy do tematu. Po wykonaniu tej czynności te rekordy zostaną zatwierdzone w kontenerze usługi Azure Cosmos DB for NoSQL zamapowanym na temat (produkty).

{"id": "0ac8b014-c3f4-4db0-8a1f-434bab460938", "name": "handlebar", "categoryId": "78148556-4e84-44be-abae-9755dde9c9e3"}
{"id": "54ba00da-50cf-44d8-b122-1d18bd1db400", "name": "handlebar", "categoryId": "eb642a5e-0c6f-4c83-b96b-bb2903b85e59"}
{"id": "381dde84-e6c2-4583-b66c-e4a4116f7d6e", "name": "handlebar", "categoryId": "cf8ae707-6d74-4563-831a-06e15a70a0dc"}

Odczyt z usługi Azure Cosmos DB

Łącznik źródłowy można utworzyć na platformie Kafka Connect przy użyciu obiektu konfiguracji JSON. W poniższej przykładowej konfiguracji większość właściwości powinna pozostać bez zmian, ale pamiętaj, aby zmienić następujące wartości:

{
  "name": "cosmosdb-source-connector",
  "config": {
    "connector.class": "com.azure.cosmos.kafka.connect.source.CosmosDBSourceConnector",
    "tasks.max": "1",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "connect.cosmos.task.poll.interval": "100",
    "connect.cosmos.connection.endpoint": "<cosmos-endpoint>",
    "connect.cosmos.master.key": "<cosmos-key>",
    "connect.cosmos.databasename": "<cosmos-database>",
    "connect.cosmos.containers.topicmap": "<kafka-topic>#<cosmos-container>",
    "connect.cosmos.offset.useLatest": false,
    "value.converter.schemas.enable": "false",
    "key.converter.schemas.enable": "false"
  }
}

Przykładem ilustracyjnym jest użycie tej przykładowej tabeli konfiguracji:

Oto przykładowy plik konfiguracji:

{
  "name": "cosmosdb-source-connector",
  "config": {
    "connector.class": "com.azure.cosmos.kafka.connect.source.CosmosDBSourceConnector",
    "tasks.max": "1",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "connect.cosmos.task.poll.interval": "100",
    "connect.cosmos.connection.endpoint": "https://dp420.documents.azure.com:443/",
    "connect.cosmos.master.key": "C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==",
    "connect.cosmos.databasename": "cosmicworks",
    "connect.cosmos.containers.topicmap": "prodlistener#products",
    "connect.cosmos.offset.useLatest": false,
    "value.converter.schemas.enable": "false",
    "key.converter.schemas.enable": "false"
  }
}

Po skonfigurowaniu dane ze źródła zmian usługi Azure Cosmos DB zostaną opublikowane w temacie platformy Kafka.