Creación de una aplicación Java que usa Azure Cosmos DB para NoSQL y procesador de fuente de cambios

Azure Cosmos DB es un servicio de base de datos NoSQL totalmente administrado proporcionado por Microsoft. Permite crear aplicaciones distribuidas globalmente y altamente escalables con facilidad. Esta guía paso a paso le guía por el proceso de creación de una aplicación Java que usa la base de datos de Azure Cosmos DB para NoSQL e implementa el procesador de fuente de cambios para el procesamiento de datos en tiempo real. La aplicación Java se comunica con Azure Cosmos DB para NoSQL mediante el SDK de Java v4 de Azure Cosmos DB.

Prerrequisitos

• Cuenta de Azure Cosmos DB: también puede crearla desde Azure Portal o usar el emulador de Azure Cosmos DB .

• Entorno de desarrollo de Java: asegúrese de que tiene el Kit de desarrollo de Java (JDK) instalado en la máquina con al menos 8 versiones.

• SDK de Java de Azure Cosmos DB V4: proporciona las características necesarias para interactuar con Azure Cosmos DB.

Contexto

La fuente de cambios de Azure Cosmos DB proporciona una interfaz controlada por eventos para desencadenar acciones en respuesta a la inserción de documentos que tiene muchos usos.

El trabajo de administrar eventos del flujo de cambios se gestiona principalmente mediante la biblioteca del Procesador de flujo de cambios integrada en el SDK. Esta biblioteca es lo suficientemente potente como para distribuir eventos de flujo de cambios entre varios trabajadores, si se desea. Lo único que tiene que hacer es proporcionar a la biblioteca de fuentes de cambios una devolución de llamada.

Este sencillo ejemplo de aplicación Java muestra el procesamiento de datos en tiempo real con Azure Cosmos DB y el procesador de fuente de cambios. La aplicación inserta documentos de ejemplo en un "contenedor de fuentes" para simular un flujo de datos. El procesador de fuente de cambios, enlazado al contenedor de fuentes, procesa los cambios entrantes y registra el contenido del documento. El procesador administra automáticamente las concesiones para el procesamiento paralelo.

Código fuente

Puede clonar el repositorio de ejemplo del SDK y encontrar este ejemplo en SampleChangeFeedProcessor.java:

git clone https://github.com/Azure-Samples/azure-cosmos-java-sql-api-samples.git
cd azure-cosmos-java-sql-api-sample/src/main/java/com/azure/cosmos/examples/changefeed/

Walkthrough

1. Configura el ChangeFeedProcessorOptions en una aplicación Java mediante Azure Cosmos DB y Azure Cosmos DB Java SDK V4. ChangeFeedProcessorOptions proporciona ajustes esenciales para controlar el comportamiento del procesador de seguimiento de cambios durante el procesamiento de datos.

   options = new ChangeFeedProcessorOptions();
   options.setStartFromBeginning(false);
   options.setLeasePrefix("myChangeFeedDeploymentUnit");
   options.setFeedPollDelay(Duration.ofSeconds(5));
   options.setFeedPollThroughputControlConfig(throughputControlGroupConfig);
   

2. Inicialice ChangeFeedProcessor con configuraciones pertinentes, incluido el nombre de host, el contenedor de fuentes, el contenedor de concesión y la lógica de control de datos. El método start() inicia el procesamiento de datos, lo que permite el procesamiento simultáneo y en tiempo real de los cambios de datos entrantes desde el contenedor de fuentes.

   logger.info("Start Change Feed Processor on worker (handles changes asynchronously)");
   ChangeFeedProcessor changeFeedProcessorInstance = new ChangeFeedProcessorBuilder()
       .hostName("SampleHost_1")
       .feedContainer(feedContainer)
       .leaseContainer(leaseContainer)
       .handleChanges(handleChanges())
       .options(options)
       .buildChangeFeedProcessor();
   changeFeedProcessorInstance.start()
                              .subscribeOn(Schedulers.boundedElastic())
                              .subscribe();
   

3. Especifique que el delegado controla los cambios de datos entrantes mediante el handleChanges() método . El método procesa los documentos JsonNode recibidos del flujo de cambios. Como desarrollador, tiene dos opciones para gestionar el documento JsonNode proporcionado por el Change Feed. Una opción es operar en el documento en forma de JsonNode. Esto es excelente especialmente si no tiene un único modelo de datos uniforme para todos los documentos. La segunda opción: transforme JsonNode en un POJO que tenga la misma estructura que JsonNode. A continuación, puede operar con el POJO.

   private static Consumer<List<JsonNode>> handleChanges() {
       return (List<JsonNode> docs) -> {
           logger.info("Start handleChanges()");
   
           for (JsonNode document : docs) {
               try {
                   //Change Feed hands the document to you in the form of a JsonNode
                   //As a developer you have two options for handling the JsonNode document provided to you by Change Feed
                   //One option is to operate on the document in the form of a JsonNode, as shown below. This is great
                   //especially if you do not have a single uniform data model for all documents.
                   logger.info("Document received: " + OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                           .writeValueAsString(document));
   
                   //You can also transform the JsonNode to a POJO having the same structure as the JsonNode,
                   //as shown below. Then you can operate on the POJO.
                   CustomPOJO2 pojo_doc = OBJECT_MAPPER.treeToValue(document, CustomPOJO2.class);
                   logger.info("id: " + pojo_doc.getId());
   
               } catch (JsonProcessingException e) {
                   e.printStackTrace();
               }
           }
           isWorkCompleted = true;
           logger.info("End handleChanges()");
   
       };
   }
   

4. Compile y ejecute la aplicación Java. La aplicación inicia el procesador de cambios, inserta documentos de ejemplo en el contenedor de cambios y procesa los cambios entrantes.

Conclusión

En esta guía, ha aprendido a crear una aplicación Java mediante el SDK de Java de Azure Cosmos DB V4 que usa la base de datos de Azure Cosmos DB para NoSQL y usa el procesador de fuente de cambios para el procesamiento de datos en tiempo real. Puede ampliar esta aplicación para controlar casos de uso más complejos y crear aplicaciones sólidas, escalables y distribuidas globalmente mediante Azure Cosmos DB.

Recursos adicionales

• Azure Cosmos DB Java SDK V4
• Ejemplos adicionales en GitHub

Pasos siguientes

Ahora puede continuar para obtener más información sobre el estimador de fuente de cambios en los siguientes artículos:

• Uso del calculador de la fuente de cambios