El ciclo de los datos, de la generación al uso.

Trabajar con grandes cantidades de datos es complicado. Muchas empresas están incorporando Map/Reduce, básicamente su implementación libre, Hadoop para el tratamiento de sus datos. El ecosistema de herramientas de Hadoop va creciendo también, y cada vez es más complejo, ya que tiene que solucionar muchos casos, pero aún le falta alguna pieza para completar el ciclo total de los datos.

Con "ciclo total de los datos" me refiero a las diferentes fases del proceso, en Hadoop normalmente:

1. Ingestión de datos en el sistema de ficheros: Hadoop trabaja sobre DFS (normalmente), así pues tenemos que enviar los datos al clúster.
2. Tratamiento de datos: Operaciones Map/Reduce en Hadoop.
3. Visualización o acceder a los datos: Una vez tenemos los resultados de los trabajos, hay que sacar los datos del DFS para poder presentarlos.

El primer punto es bastante interesante. Hadoop es útil cuando trabajamos con muchos datos (>Gbs), ahora bien, los datos deben estar en el DFS, esto puede ser un problema. El framework ofrece diferentes comandas para insertar ficheros, en mi caso un poco de scripting combinado con estos comandos siempre ha sido suficiente. Aunque estoy empezando a considerar usar Flume, un proyecto de Cloudera muy interesante que inserta datos en el sistema de ficheros distribuido como si fueran Pipes de Unix.

El segundo punto es bastante sencillo, ya que es el nucleo de Hadoop. Una vez finalizado estamos en el tercer punto, algunas veces el volumen de datos de salida será mucho inferior respecto a la entrada, a veces similar, y otras muy mayor.

Lógicamente nos interesan estos datos de salida, y la mayoría de veces necesitaremos sacarlos del DFS para poderlos estudiar, visualizar, o como es mi caso, servir a través de un servicio web.

Por qué hay que sacar los datos del DFS? por qué no los puedo usar desde allí?

Básicamente el problema viene del diseño del DFS en sí mismo, el sistema de ficheros está optimizado para los trabajos de Hadoop, que leen sequencialmente todo el fichero, esto significa que el sistema es muy lento leyendo posiciones aleatorias en un fichero. A efectos prácticos esto significa que haya mucha latencia y que no sea factible servir datos de forma rápida.

La solución más viable si no queremos (o no podemos) mover los datos es usar HBase, aunque de momento no es considerado como la mejor opción para servir las peticiones de una página web con un volumen importante de datos, aunque la comunidad está poniendo muchos esfuerzos para mejorar este tipo de rendimiento (lectura en posiciones aleatorias de ficheros) así como la estabilidad. Pero bueno, esto es otra historia.

Así pues, qué se hace para solucionar este tercer punto?

Lo más común es volcar los contenidos de los ficheros de DFS a una base de datos o a una K/V Store. Aunque tampoco es trivial.

El primer caso, sacar los ficheros y insertar el contenido a una RDBMS,  caso plantea unos problemas bastante interesantes, básicamente se pueden hacer dos cosas:

• Meterle caña a la base de datos: Una vez tenemos los ficheros del DFS se pueden convertir a SQL, luego se sube este fichero a la base de datos (con un copy), se crea una tabla nueva y se hace un swap. El problema? pues que si la nueva tabla ocupa 90 Gbytes y lo tienes que hacer cada día a lo mejor el administrador de la base de datos te viene chillando como un poseso (comprobado).
• Insertar sólo las deltas: Esta solución puede parecer más diplomática. Se trata de calcular (si se puede) las diferencias que hay entre la salida de Hadoop con el contenido de la base de datos, el resultado será una bateria de inserts, otra de updates y otra de deletes. Esta solución no es senzilla programáticamente y tiene el problema que hay que volcar los contenidos de la base de datos a Hadoop para que se pueda ver que ha cambiado. 

Nótese que en ningún momento estoy recomendando hacer las operaciones contra la base de datos desde los reducers de Hadoop, a no ser que queramos hacer un DDoS a nuestra base de datos (o que odiemos al DBA).

El segundo caso (utilizar una K/V Store en vez de una base de datos relacional) dependerá mucho de la K/V Store que utilizemos, pero la idea es la misma, podemos intentar actualizar los datos desde Hadoop y arriesgarnos a que se caiga todo o crear un fichero para que la K/V lo lea, en este caso jugamos un poco con ventaja ya que el movimiento NoSQL ha ido muy ligado al movimento Hadoop. Esto significa que muchas bases de datos no relacionales ya tienen un "conector". 

Algunos ejemplos:

• Apache Cassandra: Se puede utilizar un OutputFormat especial para que la salida de los jobs de Hadoop sea totalmente compatible con los ficheros de almacenamiento que usa Cassandra, asi pues, sólo se trata de crear estos ficheros, enviarlos a los nodos de Cassandra para que los lean. Feature un poco experimental a día de hoy.
• MongoDB: También hay un OutputFormat , en este caso escribe el resultado del tratamiento en Hadoop en BSON. El formato no forma parte de la distribución de MongoDB a día de hoy, está en un proyecto en github (no he podido probarlo aún, pero me muero de ganas). 
• Voldemort: Fue creada por los chicos de LinkedIn justamente para solucionar este problema, tiene soporte de Hadoop por defecto.
• HBase: Aunque he dicho que no es la mejor solución para servir datos online cabe decir que tiene soporte de leer ficheros generados por Hadoop por defecto.

Recientemente he estado trabajando mucho en este tema y he estado modificando un servicio web para solucionar en parte este problema. Voy a presentarlo en la séptima reunión del HUGUK el próximo 7 de noviembre (creo), espero poder colgar las transparencias y hacer un post aquí cuando lo tenga todo listo.

Mientras, hay alguien que tenga alguna sugerencia para solucionar todo este follón ?