El Ecosistema Hadoop.

Hadoop es sin duda uno de los proyectos que más acogida últimamente ha tenido entre la comunidad de software libre y es también uno de los responsables del auge de la ciencia de datos. 
Hadoop es un framework, aunque hoy en día se puede hablar de todo un ecosistema de proyectos alrededor del nucleo, vamos a intentar explicar en este post los distintos componentes de forma senzilla: 


vamos a empezar por el principio, 


Hadoop
Con Hadoop nos referimos a la parte central del sistema, que es el sistema de ficheros distribuido (HDFS) y el motor de Map/Reduce. Si echamos un vistazo a la página del proyecto veremos que hay 3 subproyectos, calma y tranquilidad!, a efectos prácticos van totalmente juntos y es una separación que se hizo por temas de código fuente:

• Hadoop MapReduce.
• HDFS.
• Hadoop - Commons.

Una vez aclarado el tema de la separación. Qué es Hadoop? Pues es la combinación de:

• un sistema de ficheros distribuido (diferentes discos duros en diferentes máquines se ven como uno solo) (HDFS)
• de un modelo de programación que se llama Map/Reduce, con su API en Java (Hadoop MapReduce)
• y un componente software que ejecuta los programas que hemos programado con la API en un cluster, usando asi el sistema de ficheros distribuido. 
• y una serie de classes comunas (Hadoop - Commons)

Esto es lo básico que debemos instalar en el clúster para obtener la funcionalidad. De este modo podemos meter datos en el sistema de ficheros y tratarlos con nuestros programas programados con el API de Hadoop. 

Vamos a ver qué más hay:

HBase

Una vez tenemos los datos procesados en Hadoop seguramente nos interesará tenerlos fácilmente accesibles para diferentes tipos de consultas, algo así como una base de datos con los resultados de Hadoop. Aquí es donde entra HBase.

HBase es una implementación libre del Bigtable de Google, es una de las famosas bases de datos NoSql. Como el nombre indica, no usa Sql, sinó que tiene una API, además el model de datos no es tabular, sinó que está basado en columnas de n dimensiones. Ya hablé de este modelo de datos en un post anterior.

A efectors prácticos, se tratan los datos en Hadoop mediante Map/Reduce y se escriben en HBase (importante saber que HBase está totalmente integrado con Hadoop, y que por lo tanto, HBase entiende perfectamente el sistema de ficheros de Hadoop asi como el formato de los ficheros), que puede estar en el mismo clúster que Hadoop o en otro, para que otras aplicaciones puedan utilizar los datos en tiempo real.

Zookeeper
Tanto HBase como Hadoop son sistemas distribuidos en los que el sistema debe controlar y saber en cada momento lo que está pasando en procesos que están en otras máquinas. Si hemos programado alguna vez este tipo de sistemas sabremos que eso no es una tarea fácil, nos vamos a encontrar todo tipos de problemas. 

Zookeeper aparece justamente para solucionarlos. Igual que HBase es la implementación libre de un paper de google, en este caso de Google Chubby. 

Básicamente es un sistema de locks distribuido y de consenso (utilizando Paxos), no hace mucho los ingenieros de LinkedIn lo usaron para un sistema interno y hicieron un post muy interesante, en el que utilizaron la siguiente definición, muy acertada en mi opinión: 

 ZK has a very simple, file system like API. One can create a path, set the value of a path, read the value of a path, delete a path, and list the children of a path. ZK does a couple of more interesting things: 

(a) one can register a watcher on a path and get notified when the children of a path or the value of a path is changed, 

(b) a path can be created as ephemeral, which means that if the client that created the path is gone, the path is automatically removed by the ZK server.

de: http://sna-projects.com/blog/2010/08/zookeeper-experience/

Asi que ya tenemos otra pieza del puzzle, si hay que sincronizar sistemas distribuidos (que utilizen o no Hadoop), podemos utilizar ZooKeeper.

Ahora toca el turno de hive, pero atención porqué después viene pig y aunque vamos a ver que son bastante diferentes, los dos proyectos pretenden solucionar un problema muy similar:



Hive
La principal incomodidad que tiene Hadoop es que para hacer una simple consulta de datos hay que escribir un programa completo en Java, y esto, no es rápido, es un poco difícil de debuggear y puede llegar a ser frustrante. 

Pongamos el caso en que tenemos 500 Gb de datos del tipo: 

Key: Integer
Value: Objecto(Id: Integer, valor1: Integer, valor2: Integer)

y que en un momento dado queremos un listado de todos los pares Key/Value dónde valor2 > 500. 

Si los datos estuvieran guardados en una base de datos podriamos hacer una senzilla consulta Sql con un where. En Hadoop tocaría escribir un programa en Java. Pues bien, Hive al rescate. 

Se trata básicamente de una infraestructura de data warehousing encima de Hadoop. Y esto que significa? pues básicamente se trata de crear unos metadatos encima de los directorios de HDFS describiendo el formato de los ficheros, asi pues creamos una estructura tabular virtual encima de Hadoop. Hive también tiene un intérprete de comandas sql, por lo tanto podemos escribir una consulta sql de toda la vida en la command line, luego hive coge el sql, lo convierte automáticamente en jobs Map/Reduce y estos son ejecutados de forma transparente en el clúster como jobs normales. 


Si quereis más información, anteriormente hice un post hablando de hive con más detalle.


Pig
Otro modo de ver el anterior problema es: en vez de escribir un programa en Java para hacer una consulta de unos datos, voy a hacerlo con nserta aquí tu lenguaje de scripting favorito>. Hadoop tiene como parte del Hadoop Core una opción de streaming. Básicamente te permite especificar como parámetro dos scripts (uno que hace de mapper y otro que hace de reducer) que leen y escriben por la entrada y salida estándar. Esto es práctico pero te limita a unos programas muy sencillos y poco aplicables a la vida real. 

Pues bien, Pig es un lenguage de programación (tipo scripting) para Hadoop, y por lo que estoy viendo parece que tiene una muy buena aceptación en la comunidad Hadoop (pero que muy buena aceptación).

Avro
Una de los puntos fuertes del ecosistema Hadoop es que facilita trabajar con grandes cantidades de datos.  Estos datos están guardados en el HDFS, pero cómo? pues se pueden guardar en diferentes formatos, el más simple es el de texto plano, pero normalmente se recomienda usar serialización. 


Qué es la serialización? pues se trata de como almacenamos la información internamente de un objeto a disco (o memoria), podemos leer más en la wikipedia. Para Hadoop lo más interesante son los formatos binarios, que son los más rápidos y comprimidos, el problema es que un humano no los puede leer, pero en este caso no nos importa. 


Tenemos diferentes formatos: 

• Protocol Buffers, creado por Google.
• Thrift, creado por Facebook.

y otros dos, creados por los programadors de Hadoop:

• RecordIO: Hasta hace pronto el formato de serialización de facto en Hadoop, bastante interesante en su implementación, pero con algunos problemas en temas de mantenimiento a largo plazo (los objetos generados son estáticos, esto significa que si modificamos las características del objeto (introducimos un nuevo campo) el objeto nuevo será incompatible con el viejo). Si echamos una ojeada a la documentación veremos que está deprecated by... si señor:
• Avro.

Tiene unas características muy interesantes, básicamente: 

• Permite almacenar estructuras de datos simples y complejas (desde números hasta objectos con listas y maps)
• Es compacto, rápido y binario. (Esto no es mucha novedad)
• Tiene un format de fichero propio.
• Tiene una especificación de RPC. (como Thrift)
• Y lo más interesante: El esquema de serialización (el índice que dice que tipo de datos está en cada posición del fichero) está incluido en el fichero en si. Esto permite que la generación de código por parte del precompilador sea opcional y lo que es más importante, hace que podamos canviar la estructura interna de los objectos almacenados sin perder la compatibilidad con ficheros viejos.

Por lo tanto, si trabajais con Hadoop, a serializar los datos con Avro desde ya.

Flume
The "new kid in town". Flume fue liberado por Cloudera no hace ni medio año y ya ha tenido una gran adopción. 

La idea es muy sencilla, se trata de una pipe al HDFS. 
Lógicamente para trabajar con Hadoop debemos tener los datos en el HDFS, y tenemos diferentes herramientas para poner los ficheros allí, el problema es que cuando tenemos diferentes fuentes de datos (el syslog de muchos servidores web por ejemplo) hay que construir una infraestructura que se asegure que los logs se van subiendo periódicamente y que no falte ninguno. Pues bien, Flume es exactamente esto. Además utiliza ZooKeeper, por lo tanto es:

•  tolerante a fallos ( si un nodo cae de la red, cuando se vuelva a conectar sabrá que tiene que enviar de nuevo ), 
• tiene configuración distribuida (puede reconfigurar los nodos cliente a partir de un nodo máster) 

Y no se acaba aquí! hay más, mucho más en Hadoop, pero creo que estos son los proyectos más interesantes y que estás más ligados a Hadoop.

Echáis en falta alguno?

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 ?

Linked Data y Government 2.0

Hace poco se publicó la página web http://www.legislation.gov.uk/ del gobierno inglés. Se trata (como podemos leer en http://data.gov.uk/blog/legislationgovuk-api) de una lugar en dónde se pueden consultar todas las leyes que afectan a Inglaterra, Gales, Irlanda del Norte y Escocia así como sus cambios a través del tiempo.

Qué gracia tiene?  pues dos:

1. Incluye una API para que terceros desarrollen aplicaciones y
2. Cada ley está identificada únicamente por una URL, y está descrita con RDF, por tanto, es un resource de Linked Data.

A efectos prácticos (técnicos) esto significa que una aplicación con conocimiento semántico pueda interpretar las leyes y que estas sean únicas y referenciables siempre, un requisito de la Web de Datos ( o 3.0).

A efectos prácticos (no técnicos) esto significa que el gobierno británico continúa adelante con su apuesta por abrir sus datos para desarrolladores y colabora a la construcción de la Web Semántica.
Recordemos que el gobierno británico, juntamente con el americano, son los que están apostando más con lo que se llama el Gobierno 2.0 (o Open Government, o e-Gobernment, o ...). La página http://data.gov.uk/ es simplemente espectacular.

Y en España?
En España el tema no está nada mal, podría ser peor lógicamente, pero tenemos algunos gobiernos autonómicos (como el basco) que están publicando sus datos y Institutos de Estadística que también lo hacen (yo mismo hice una implementación en Java de la API del Idescat). Y también aparecen iniciativas como las del concurso Abredatos 2010, de las que salieron proyectos muy interesantes.

Durante el concurso Abredatos, participé con un amigo y la principal dificultad (bueno, la segunda, nuestra principal dificultad fue el CSS de la página) fue sacar los datos de diferentes páginas. El INE ofrece datos, pero desgraciadamente es muy difícil parsearlos, los datos están pensados para visualizarlos desde la web, o para descargarlos con excel (cada fichero con un formato diferente  por cierto).

Es en este sentido que la apertura de datos en España podría mejorar mucho,  mis requerimientos para una apertura de datos real serían:

1. Abrir los datos con una licencia no restrictiva.
2. Publicarlos via API
3. Describir los datos semánticamente.

En la conferencia NoSql - EU 2010 vi una presentación de unos programadores de The Guardian, en ella explicaban sus experiencias con bases de datos no relacionales, uno de los ejemplos (con Redis) se trató del escándalo de los gastos de los ministros en Inglaterra.

El gobierno publicó en pdf miles de documentos con las facturas de los ministros. Lógicamente se tardaría mucho en poder revisarlos uno por uno, lo que hiciero fue crear una página web dónde los ciudadanos podían ver facturas aleatorias, y seguidamente clasificar la factura como "escandalosa" (para que un reportero del diario la revisara) o como "aceptable". He aquí un muy buen ejemplo de lo que se puede ganar la ciudadanía con las nuevas tecnologías y la apertura de datos y lo que pueden perder algunas personas que están muy bien acostumbradas.

Por cierto, esta es la presentación de The Guardian (el tema de los gastos empieza en la página 46):