Recientemente vi una pregunta en nuestros foros donde alguien quería mover algunos datos de CouchDB a Couchbase. Como suelo ayudar a los amigos que lo necesitan cuando se están mudando, pensé que podría ayudar. Mudarse requiere preparación, especialmente con muebles grandes como Couch :D
Perdón por la metáfora, ahora os explico cómo lo he hecho. Mi objetivo es mover datos de CouchDB a Couchbase. Así que la primera pregunta es ¿cómo sacar los datos de CouchDB? Hay varias opciones disponibles y la más sencilla para mí fue utilizar la API REST. Si usas el endpoint _all_docs con el parámetro include_doc en true, obtendrás todos los documentos. Esto es exactamente lo que necesito.
Ahora, en lugar de descargar la respuesta REST directamente, puedo utilizar la API de Java 8 stream. Y puesto que voy a utilizar RxJava como parte de nuestro SDK, necesito envolver ese flujo en un observable. En realidad es bastante simple:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
URL url = nuevo URL(couchDBRequest); HttpURLConnection conn = (HttpURLConnection) url.openConnection(); conn.setRequestMethod("GET"); conn.setRequestProperty("Acepte", "aplicación/json"); //asumo que este va a ser un archivo grande... conn.setReadTimeout(0); si (conn.getResponseCode() != 200) { tirar nuevo RuntimeException("Fallido : HTTP error código : " + conn.getResponseCode()); } BufferedReader inp2 = nuevo BufferedReader(nuevo InputStreamReader(conn.getInputStream())); final largo[] totalFilas = nuevo largo[2]; int cuente = Observable.de(inp2.líneas()::iterador).cuente().toBlocking().solo(); |
Con esto tengo un observable que emite documentos de CouchDB. El siguiente paso es, por supuesto, tomar estos documentos y enviarlos a Couchbase. Primero usaré un flatMap para parsear cada línea de la respuesta. Es fácil de hacer porque cada línea de la respuesta contiene un documento como puedes ver:
|
1 2 3 4 5 6 |
{"total_rows":2,"offset":0,"filas":[ {"id":"f750a4273b48b6c1146fe4ead1000c1b","llave":"f750a4273b48b6c1146fe4ead1000c1b","valor":{"rev":"2-bea53b374bf5a427ab15245bc029cac0"},"doc":{"_id":"f750a4273b48b6c1146fe4ead1000c1b","_rev":"2-bea53b374bf5a427ab15245bc029cac0","título:"Un título"}}, {"id":"f750a4273b48b6c1146fe4ead1000c24","llave":"f750a4273b48b6c1146fe4ead1000c24","valor":{"rev":"1-9baa68f46c29940ad7a6d57ae1a04002"},"doc":{"_id":"f750a4273b48b6c1146fe4ead1000c24","_rev":"1-9baa68f46c29940ad7a6d57ae1a04002","título:"Otro título"}} ]} |
Tengo que manejar la primera y la última línea por separado. En la primera línea obtengo el total_rows y la información de offset. En la última línea no tengo que hacer nada. Simplemente devuelvo Observable.empty() para esas dos líneas ya que no tengo nada que alimentar al siguiente operador. Todas las demás líneas contienen una fila editada por CouchDB. Cada una de estas filas contiene un documento JSON que podemos envolver en un JsonNode.
El siguiente operador también es un flatMap. Aquí extraigo la clave del documento y su contenido como String. Como tengo un objeto Json como String, no necesito ningún tipo de mapeo usando Jackson o similares. Puedo utilizar directamente un RawJsonDocument. Una vez que tengo el RawJsonDocument puedo importarlo a Couchbase. Para ello utilizo el método upsert. Es algo así como "sin preguntas". No nos importa si el documento existe o no: si no existe será creado, mientras que si existe será reemplazado. Puede que este no sea el comportamiento que usted desea, pero es el más simple para este escenario, ya que significa que no tengo que manejar errores cuando una clave ya existe.
Asigno un tiempo de espera de 500 milisegundos a la operación, ya que realmente no debería tardar mucho. Luego uso el RetryBuilder. Es un buen ayudante añadido por Simon Basle para gestionar fácilmente los reintentos en caso de error. Aquí reintento hasta 100 veces si obtengo una RequestCancelledException. Añado un retardo arbitrario de 31 segundos antes de cada reintento. Hago lo mismo para la TemporaryFailureException y la BackpressureException. Aquí utilizo un retardo de 100 milisegundos.
Luego uso doOnError y doOnNext para registrar la clave del documento ya sea en el archivo de registro de éxito o en el archivo de registro de error. Los métodos doOnX no cambian la semántica central de tu flujo (como lo hace la transformación de datos, por ejemplo) sino que agregan algún comportamiento adicional, "efectos secundarios". Aquí escribo una cadena en un fichero de registro usando FileWriter. Desafortunadamente, esto es síncrono y bloqueante. Podría cambiar esto para usar un logger asíncrono en su lugar.
Luego uso onErrorResumeNext para asegurarme de que la importación continúa aunque haya errores. Por último utilizo count().toBlocking().single() para saber cuántas inserciones he hecho en Couchbase. Comparo el resultado con el número de total_rows al final.
Al final el código queda así:
Todavía hay muchas mejoras posibles. Como tener una opción de configuración para elegir el nivel de consistencia que se desea durante la importación (opciones PersistTo y ReplicateTo). También estaría bien dar como entrada al script el registro de errores que contiene sólo las claves de los documentos que no se importaron. De esta forma se puede repetir una importación sólo de los documentos con error.
En cualquier caso, espero que te sirva de ayuda.