Getting started with Kafka and Couchbase as an endpoint

Couchbase es genial como fuente para Apache Kafka usando el conector DCP.
Sin embargo, también es estupendo como punto final para digerir datos, ya que es rápido, primero memoria y almacenamiento fiable.

En esta entrada del blog te mostraré cómo construir una aplicación Java simple para un productor y un consumidor que guardan los mensajes publicados de Kafka en Couchbase.

Asumo aquí, que ya tienes un cluster Kafka (incluso si es un cluster de un solo nodo). Si no es así, intenta seguir esa guía de instalación.

Este entorno de blog tiene 4 partes:

Productor de Kafka
Cola de Apache Kafka
Consumidor Kafka
Servidor Couchbase

Productor

Necesitamos el productor para enviar mensajes a nuestra cola.

En la cola, esos mensajes están siendo digeridos y cada aplicación que se suscribió al tema - puede leer esos mensajes.

La fuente de nuestros mensajes será un archivo JSON ficticio que he creado usando Mockaroo, que dividiremos y enviaremos a la cola.

Nuestros datos JSON de muestra tienen un aspecto similar a:

{   
   "id":1,  
   "gender":"Female",  
   "first_name":"Jane",  
   "last_name":"Holmes",  
   "email":"jholmes0@myspace.com",  
   "ip_address":"230.49.112.20",  
   "city":"Houston"  
 }

{

"id":1,

"género":"Mujer",

"nombre":"Jane",

"apellido":"Holmes",

"email":"jholmes0@myspace.com",

"dirección_ip":"230.49.112.20",

"ciudad":"Houston"

}

El código del productor:

import com.fasterxml.jackson.databind.JsonNode;  
 import com.fasterxml.jackson.databind.ObjectMapper;  
 import com.fasterxml.jackson.databind.node.ArrayNode;  
 import org.apache.kafka.clients.producer.KafkaProducer;  
 import org.apache.kafka.clients.producer.ProducerConfig;  
 import org.apache.kafka.clients.producer.ProducerRecord;  
 import org.apache.kafka.clients.producer.RecordMetadata;  
   
 import java.io.File;  
 import java.nio.charset.Charset;  
 import java.nio.file.Files;  
 import java.nio.file.Paths;  
 import java.util.ArrayList;  
 import java.util.HashMap;  
 import java.util.List;  
 import java.util.Map;  
 import java.util.concurrent.Future;  
   
   
 public class KafkaSimpleProducer {  
   public static void main(String[] args) throws Exception {  
     Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();  
     config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");  
     config.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
     config.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
     KafkaProducer&lt;String, String&gt; producer = new KafkaProducer&lt;String, String&gt;(config);  
   
     File input = new File("sampleJsonData.json");  
     byte[] encoded = Files.readAllBytes(Paths.get(input.getPath()  ));  
   
     String jsons = new String(encoded, Charset.defaultCharset());  
     System.out.println("Splitting file to jsons....");  
   
     List splittedJsons = split(jsons);  

     System.out.println("Converting to JsonDocuments....");  
   
     int docCount = splittedJsons.size();  
   
     System.out.println("Number of documents is: " + docCount );  
   
     System.out.println("Starting sending msg to kafka....");  
     int count = 0;  
     for ( String doc : splittedJsons) {  
       System.out.println("sending msg...." + count);  
       ProducerRecord&lt;String,String&gt; record = new ProducerRecord&lt;&gt;( "couchbaseTopic", doc );  
       Future meta = producer.send(record);  
       System.out.println("msg sent...." + count);  
   
       count++;  
     }  
   
     System.out.println("Total of " + count + " messages sent");  
   
     producer.close();  
   }  

   public static List split(String jsonArray) throws Exception {  
     List splittedJsonElements = new ArrayList();  
     ObjectMapper jsonMapper = new ObjectMapper();  
     JsonNode jsonNode = jsonMapper.readTree(jsonArray);  
   
     if (jsonNode.isArray()) {  
       ArrayNode arrayNode = (ArrayNode) jsonNode;  
       for (int i = 0; i &lt; arrayNode.size(); i++) {  
         JsonNode individualElement = arrayNode.get(i);  
         splittedJsonElements.add(individualElement.toString());  
       }  
     }  
     return splittedJsonElements;  
   }  
 }

importar com.fasterxml.jackson.databind.JsonNode;

importar com.fasterxml.jackson.databind.ObjectMapper;

importar com.fasterxml.jackson.databind.nodo.ArrayNode;

importar org.apache.kafka.clientes.productor.KafkaProducer;

importar org.apache.kafka.clientes.productor.ProducerConfig;

importar org.apache.kafka.clientes.productor.ProductorRegistro;

importar org.apache.kafka.clientes.productor.RecordMetadata;

importar java.io.Archivo;

importar java.nio.conjunto de caracteres.Juego de caracteres;

importar java.nio.archivo.Archivos;

importar java.nio.archivo.Caminos;

importar java.util.ArrayList;

importar java.util.HashMap;

importar java.util.Lista;

importar java.util.Mapa;

importar java.util.concurrente.Futuro;

público clase KafkaSimpleProducer {

público estático void principal(Cadena[] args) lanza Excepción {

Mapa<Cadena, Objeto> config = nuevo HashMap<>();

config.poner(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

config.poner("valor.serializador", "org.apache.kafka.common.serialization.StringSerializer");

config.poner("clave.serializador", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<Cadena, Cadena> productor = nuevo KafkaProducer<Cadena, Cadena>(config);

Archivo entrada = nuevo Archivo("sampleJsonData.json");

byte[] codificado = Archivos.readAllBytes(Caminos.consiga(entrada.getPath() ));

Cadena jsons = nuevo Cadena(codificado, Juego de caracteres.defaultCharset());

Sistema.fuera.println("Dividir archivo en jsons....");

Lista splittedJsons = dividir(jsons);

Sistema.fuera.println("Convirtiendo a JsonDocuments....");

int docCount = splittedJsons.talla();

Sistema.fuera.println("El número de documentos es: " + docCount );

Sistema.fuera.println("Empezando a enviar msg a kafka....");

int cuente = 0;

para ( Cadena doc : splittedJsons) {

Sistema.fuera.println("enviando msg...." + cuente);

ProductorRegistro<Cadena,Cadena> registro = nuevo ProductorRegistro<>( "couchbaseTopic", doc );

Futuro meta = productor.enviar(registro);

Sistema.fuera.println("msg sent...." + cuente);

cuente++;

}

Sistema.fuera.println("Total de " + cuente + " mensajes enviados");

productor.cerrar();

}

público estático Lista dividir(Cadena jsonArray) lanza Excepción {

Lista splittedJsonElements = nuevo ArrayList();

ObjectMapper jsonMapper = nuevo ObjectMapper();

JsonNode jsonNode = jsonMapper.readTree(jsonArray);

si (jsonNode.isArray()) {

ArrayNode arrayNodo = (ArrayNode) jsonNode;

para (int i = 0; i < arrayNodo.talla(); i++) {

JsonNode elementoindividual = arrayNodo.consiga(i);

splittedJsonElements.añada(elementoindividual.toString());

}

devolver splittedJsonElements;

}

Salida de la aplicación productora de Kafka

Consumidores

Esta es una simple, muy directa, solo obtén los mensajes de la cola, y usa el Couchbase Java SDK para insertar documentos en Couchbase. Por simplicidad, usaré el sync java SDK, pero usar el async es totalmente posible e incluso recomendable.

import com.couchbase.client.java.Bucket;  
 import com.couchbase.client.java.Cluster;  
 import com.couchbase.client.java.CouchbaseCluster;  
 import com.couchbase.client.java.document.JsonDocument;  
 import com.couchbase.client.java.document.json.JsonObject;  
 import kafka.consumer.Consumer;  
 import kafka.consumer.ConsumerConfig;  
 import kafka.consumer.KafkaStream;  
 import kafka.javaapi.consumer.ConsumerConnector;  
 import kafka.message.MessageAndMetadata;  
   
 import java.util.*;  
   
 public class KafkaSimpleConsumer {  
   public static void main(String[] args) {  
   
     Properties config = new Properties();  
     config.put("zookeeper.connect", "localhost:2181");  
     config.put("zookeeper.connectiontimeout.ms", "10000");  
     config.put("group.id", "default");  
   
     ConsumerConfig consumerConfig = new kafka.consumer.ConsumerConfig(config);  
   
     ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);  
   
     Map&lt;String, Integer&gt; topicCountMap = new HashMap&lt;&gt;();  
     topicCountMap.put("couchbaseTopic", 1);  
   
     Map&lt;String, List&lt;KafkaStream&lt;byte[], byte[]&gt;&gt;&gt; consumerMap = consumerConnector.createMessageStreams(topicCountMap);  
   
     List&lt;KafkaStream&lt;byte[], byte[]&gt;&gt; streams = consumerMap.get("couchbaseTopic");  
   
     List nodes = new ArrayList&lt;&gt;();  
     nodes.add("localhost");  
   
     Cluster cluster = CouchbaseCluster.create(nodes);  
     final Bucket bucket = cluster.openBucket("kafkaExample");  
   
     try {  
       for (final KafkaStream&lt;byte[], byte[]&gt; stream : streams) {  
         for (MessageAndMetadata&lt;byte[], byte[]&gt; msgAndMetaData : stream) {  
           String msg = convertPayloadToString(msgAndMetaData.message());  
           System.out.println(msgAndMetaData.topic() + ": " + msg);  
   
           try {  
             JsonObject doc = JsonObject.fromJson(msg);  
             String id = UUID.randomUUID().toString();  
             bucket.upsert(JsonDocument.create(id, doc));  
           } catch (Exception ex) {  
             System.out.println("Not a json object: " + ex.getMessage());  
           }  
         }  
       }  
     } catch (Exception ex) {  
       System.out.println("EXCEPTION!!!!" + ex.getMessage());  
       cluster.disconnect();  
     }  
   
     cluster.disconnect();  
   }  
   
   private static String convertPayloadToString(final byte[] message) {  
     String string = new String(message);  
     return string;  
   }  
 }

importar com.couchbase.cliente.java.Cubo;

importar com.couchbase.cliente.java.Grupo;

importar com.couchbase.cliente.java.CouchbaseCluster;

importar com.couchbase.cliente.java.documento.JsonDocument;

importar com.couchbase.cliente.java.documento.json.JsonObject;

importar kafka.consumidor.Consumidores;

importar kafka.consumidor.ConsumerConfig;

importar kafka.consumidor.KafkaStream;

importar kafka.javaapi.consumidor.ConectorConsumidor;

importar kafka.mensaje.Mensaje y metadatos;

importar java.util.*;

público clase KafkaSimpleConsumer {

público estático void principal(Cadena[] args) {

Propiedades config = nuevo Propiedades();

config.poner("zookeeper.connect", "localhost:2181");

config.poner("zookeeper.connectiontimeout.ms", "10000");

config.poner("grupo.id", "por defecto");

ConsumerConfig consumerConfig = nuevo kafka.consumidor.ConsumerConfig(config);

ConectorConsumidor consumerConnector = Consumidores.createJavaConsumerConnector(consumerConfig);

Mapa<Cadena, Entero> topicCountMap = nuevo HashMap<>();

topicCountMap.poner("couchbaseTopic", 1);

Mapa<Cadena, Lista<KafkaStream<byte[], byte[]>>> consumerMap = consumerConnector.createMessageStreams(topicCountMap);

Lista<KafkaStream<byte[], byte[]>> arroyos = consumerMap.consiga("couchbaseTopic");

Lista nodos = nuevo ArrayList<>();

nodos.añada("localhost");

Grupo grupo = CouchbaseCluster.crear(nodos);

final Cubo cubo = grupo.openBucket("kafkaExample");

pruebe {

para (final KafkaStream<byte[], byte[]> flujo : arroyos) {

para (Mensaje y metadatos<byte[], byte[]> msgAndMetaData : flujo) {

Cadena msg = convertPayloadToString(msgAndMetaData.mensaje());

Sistema.fuera.println(msgAndMetaData.tema() + ": " + msg);

pruebe {

JsonObject doc = JsonObject.fromJson(msg);

Cadena id = UUID.randomUUID().toString();

cubo.upsert(JsonDocument.crear(id, doc));

} captura (Excepción ex) {

Sistema.fuera.println("No es un objeto json: " + ex.getMessage());

}

} captura (Excepción ex) {

Sistema.fuera.println("EXCEPTION!!!!" + ex.getMessage());

grupo.desconectar();

}

grupo.desconectar();

}

privado estático Cadena convertPayloadToString(final byte[] mensaje) {

Cadena cadena = nuevo Cadena(mensaje);

devolver cadena;

}

Salida de la consola de nuestro consumidor Kafka

Servidor Couchbase

Ahora podemos ver el resultado en el servidor Couchbase.

Mira el bucket kafkaExample - Lleno con 1000 documentos.

Cubos Couchbase

Cada documento tiene un aspecto similar:

Documento de muestra

Solución sencilla en 3 partes.

Tenga en cuenta que en un entorno de producción, Producer, Consumer, Kafka o Couchbase estarán en una o más máquinas cada uno.

Código completo (incluidas las dependencias de Maven) en GitHub.

Roi.

Roi Katz, Arquitecto de soluciones, Couchbase

Platform

Self-Managed

Services

Capabilities

Why Couchbase?

Migrate to Capella

By Use Case

By Industry

By Application Need

Popular Docs

By Developer Role

Quickstart

Resource Center

About

Partnerships

Our Services

Partners: Register a Deal

Ready to register a deal with Couchbase?

Marriott

Todos los mensajes

Primeros pasos con Kafka y Couchbase como endpoint

Productor

Consumidores

Servidor Couchbase

Autor

Publicado por Roi Katz, Arquitecto de soluciones, Couchbase

Dejar una respuesta Cancelar respuesta