En este post veremos un resumen de la API de Trident que podremos ir siguiendo de forma práctica con la ayuda del proyecto github anteriormente mencionado.

Storm (a modo de resumen)

A modo de resumen, Storm es una herramienta para el proceso en tiempo de real de flujos de datos. Lo que lo hace diferente es que es una abstracción de más alto que nivel que el simple envío de mensajes (permite definir topologías como un grafo acíclico dirigido), tiene tolerancia a fallos para cada proceso y garantiza que cada mensaje en el sistema se va a procesar por lo menos una vez (at-least-once semantics).

Trident (a modo de resumen)

Trident es una abstracción interesante por encima de Storm. Además de proveer métodos de más alto nivel tipo los de Cascading, provée “mini-batches” de tuplas para 1) hacer más fácil el construir la lógica de procesamiento y 2) favorecer la persistencia eficiente de los datos, incluso con la ayuda de una API que puede dar seguridad total en este aspecto (“exactly-once” semantics).

Hay que recordar y tener en cuenta que almacenar estado en memoria en los Bolts de Storm no es totalmente tolerante a fallos: si un nodo muere, el proceso se reasignará, pero el estado no se podrá recuperar. Hay un ticket en github para ésto, pero mientrastanto lo más sensato es usar Storm para persistir cualquier estado a una base de datos, y para eso Trident es especialmente útil. Ya que si estamos manejando grandes cantidades de datos, necesitamos hacer algún tipo de encolado para no tener que actualizar la base de datos para cada uno de los mensajes que procesamos. Los “mini-batches” a los que nos hemos referido nos ayudan precisamente a ello, y Trident nos proporciona una API de agregación para ellos que nos permite persistir el resultado en cualquier base de datos.

Comenzando: each()

El punto de entrada de nuestro tutorial es la clase Skeleton. Aquí vemos el uso de la clase FakeTweetsBatchSpout que es el Spout que nos va a generar el stream de tweets falsos. Podemos cambiar el tamaño de cada batch por constructor. La primera operación que veremos es each(), que nos permite manipular cada tupla del batch ya sea filtrándola o aplicándole alguna función. Podemos implementar un filtro que filtre los tweets escritos por un cierto actor:

public static class PerActorTweetsFilter extends BaseFilter {
  String actor;

  public PerActorTweetsFilter(String actor) {
    this.actor = actor;
  }
  @Override
  public boolean isKeep(TridentTuple tuple) {
    return tuple.getString(0).equals(actor);
  }
}

Podemos añadir el filtro previo a la definición de la topología:

topology.newStream("spout", spout)
  .each(new Fields("actor", "text"), new PerActorTweetsFilter("dave"))
  .each(new Fields("actor", "text"), new Utils.PrintFilter());

Observad cómo en el selector de campos (Fields) hemos seleccionado “actor” y “text”. Las tuplas que entren en cada método each() puede que tengan muchos campos, pero éste selector nos permite seleccionar sólo un subconjunto de interés. Por lo tanto la tupla que entre en el filtro será un array de dos campos con el actor en posición 0 y el texto del tweet en posición 1. También hemos añadido otro filtro que sólo imprime por pantalla cada tupla que pasa por él. El comportamiento de esta topología es bastante predecible: filtrará todo tweet que no fue escrito por “dave”.

Veamos ahora una función ejemplo:

public static class UppercaseFunction extends BaseFunction {
  @Override
  public void execute(TridentTuple tuple, TridentCollector collector) {
    collector.emit(new Values(tuple.getString(0).toUpperCase()));
  }
}

Esta función simplemente emite toda cadena de texto en posición 0 de la tupla y la imprime en mayúsculas. Podemos añadir esta función a la definición de la topología, por ejemplo:

topology.newStream("spout", spout)
  .each(new Fields("actor", "text"), new PerActorTweetsFilter("dave"))
  .each(new Fields("text", "actor"), new UppercaseFunction(), new Fields("uppercased_text"))
  .each(new Fields("actor", "text", "uppercased_text"), new Utils.PrintFilter());

Veamos dos cosas. La primera es que el selector de campos correspondiente a la llamada de la función lo hemos invertido con respecto al de la llamada del primer filtro, ya que la función convertirá a mayúsculas la cadena que haya en la posición 0. La segunda cosa es que para la llamada de la función tenemos un descriptor de campos de salida. Siempre hemos de describir los campos que emite una función (en este caso, “uppercased_text”). Las tuplas que salgan de la función tendrán todos los campos de la tupla de entrada más los que hayamos emitido como resultado de la función (así que no podemos llamar a un campo de salida igual que uno de entrada). El código de arriba pasará a mayúsculas todos los tweets escritos por “dave” e imprimirá tanto el texto original como el texto en mayúsculas.

Como curiosidad, decir que cada llamada a each() nos permite hacer una “proyección” implícita de las tuplas seleccionando un subconjunto de sus campos, pero si por algún motivo necesitáramos hacer una proyección explícita podemos usar el método project().

Haciendo las cosas más interesantes: parallelismHint() y partitionBy()

Volvamos al ejemplo del filtro simple. ¿Qué pasa si definimos la topología de la siguiente manera?

topology.newStream("spout", spout)
  .each(new Fields("actor", "text"), new PerActorTweetsFilter("dave"))
  .parallelismHit(5)
  .each(new Fields("actor", "text"), new Utils.PrintFilter());

parallelismHint() configurará la topología hasta donde lo hayamos puesto para ser ejecutada con un cierto nivel de paralelismo. Esto no es 100% verdad, pero por ahora es una definición suficientemente buena. Para visualizar ésto mejor, podemos modificar el filtro PerActorTweetsFilter:

public static class PerActorTweetsFilter extends BaseFilter {

  private int partitionIndex;
  private String actor;

  public PerActorTweetsFilter(String actor) {
    this.actor = actor;
  }
  @Override
  public void prepare(Map conf, TridentOperationContext context) {
    this.partitionIndex = context.getPartitionIndex();
  }
  @Override
  public boolean isKeep(TridentTuple tuple) {
    boolean filter = tuple.getString(0).equals(actor);
    if(filter) {
      System.err.println("I am partition [" + partitionIndex + "] and I have kept a tweet by: " + actor);
    }
    return filter;
  }
}

Si ahora ejecutamos la topología obtendremos resultados como:

I am partition [4] and I have kept a tweet by: dave
I am partition [3] and I have kept a tweet by: dave
I am partition [0] and I have kept a tweet by: dave
I am partition [2] and I have kept a tweet by: dave
I am partition [1] and I have kept a tweet by: dave

Ésto nos dice claramente que el filtro está siendo ejecutado en paralelo por 5 procesos diferentes. También tenemos 5 Spouts generadores de tuplas (si buscamos la cadena “Open Spout instance” en los logs nos daremos cuenta de ello). ¿Qué pasa si sólo queremos 2 Spouts y 5 procesos de filtrado?

topology.newStream("spout", spout)
  .parallelismHint(2)
  .shuffle()
  .each(new Fields("actor", "text"), new PerActorTweetsFilter("dave"))
  .parallelismHint(5)
  .each(new Fields("actor", "text"), new Utils.PrintFilter());

El método shuffle() es una operación de reparticionado. Hay otras operaciones como partitionBy() o global() que también son de reparticionado. El reparticionado nos permite especificar cómo se han de enrutar las tuplas al siguiente nivel de procesamiento, así como hacer que cada nivel de procesamiento se ejecute con diferentes niveles de paralelismo. Shuffle() hace un reparticionado alteatorio mientras que partitionBy() hace un enrutado basado en “consistent hashing” sobre los campos que se han especificado. Dicho ésto, ahora ya podemos clarificar la anterior definición de parallelismHint(): ésta función aplica un cierto nivel de paralelismo a todas las operaciones anteriores a donde la hemos especificado, hasta que haya una operación de reparticionado de algún tipo.

Cambiemos shuffle() por partitionBy(new Fields(“actor”)). ¿Qué creeis que pasará?

I am partition [2] and I have kept a tweet by: dave
I am partition [2] and I have kept a tweet by: dave
I am partition [2] and I have kept a tweet by: dave
I am partition [2] and I have kept a tweet by: dave

Como hemos usado partitionBy(new Fields(“actor”)) estamos diciendo que todas las tuplas del mismo actor tienen que ir forzosamente al mismo proceso, así que no es de extrañar que ahora sólo haya un proceso ejecutando el filtrado a pesar de que la capa de procesamiento tiene paralelismo 5.

Agregación

Hemos dicho antes que Trident nos permite procesar “mini-batches” de tuplas. Una operación “batch”, por lotes, que nos viene de forma natural a la cabeza es la agregación. Trident nos provée primitivas para hacer agregaciones sobre los batches. Veamos un ejemplo:

public static class LocationAggregator extends BaseAggregator<Map<String, Integer>> {

  @Override
  public Map<String, Integer> init(Object batchId, TridentCollector collector) {
    return new HashMap<String, Integer>();
  }

  @Override
  public void aggregate(Map<String, Integer> val, TridentTuple tuple, TridentCollector collector) {
    String location = tuple.getString(0);
    val.put(location, MapUtils.getInteger(val, location, 0) + 1);
  }

  @Override
  public void complete(Map<String, Integer> val, TridentCollector collector) {
    collector.emit(new Values(val));
  }
}

Este agregador es muy simple: la idea es procesar cada batch de tuplas para obtener un mapa que cuente por localización. A través de este ejemplo vemos la interfaz “Aggregator”, observemos cómo Trident llamará al método init() al principio de cada batch, al método aggregate() para cada tupla dentro del batch y al método complete() al final de cada batch. Podemos usar la instancia “collector” en cualquier momento de este ciclo, pero en este ejemplo sólo lo usaríamos al final del batch por eficiencia: podríamos usar el resultado del agregador para actualizar una base de datos, por ejemplo.

Podemos testear ésto usando el método aggregate() de Trident. El método aggregate() es también una operación de reparticionado. Agregará todas las tuplas de un batch en un único proceso, escogido al azar. Para minimizar los datos enviados por red, y siempre que sea posible, podríamos usar la interfaz CombinerAggregator. Pero por ahora sigamos en la interfaz de más bajo nivel y probémosla:

TridentTopology topology = new TridentTopology();
topology.newStream("spout", spout)
  .aggregate(new Fields("location"), new LocationAggregator(), new Fields("location_counts"))
  .each(new Fields("location_counts"), new Utils.PrintFilter());

Obtenemos resultados como los siguientes:

[{USA=3, Spain=1, UK=1}]
[{USA=3, Spain=2}]
[{France=1, USA=4}]
[{USA=4, Spain=1}]
[{USA=5}]

Fijaros como la suma de las cuentas es siempre 5. Ésto es porque el tamaño de batch por defecto es de 5.

Pero incrementémoslo ahora a 100:

FakeTweetsBatchSpout spout = new FakeTweetsBatchSpout(100);

Ahora modifiquemos un poco la topología y tratemos de adivinar lo que pasará:

TridentTopology topology = new TridentTopology();
topology.newStream("spout", spout)
  .partitionBy(new Fields("location"))
  .partitionAggregate(new Fields("location"), new LocationAggregator(), new Fields("location_counts"))
  .parallelismHint(3)
  .each(new Fields("location_counts"), new Utils.PrintFilter());

Ahora obtenemos resultados como los siguientes:

[{France=10, Spain=5}]
[{USA=63}]
[{UK=22}]

En efecto, partitionAggregate() no es una operación de reparticionado. En vez de reparticionar, ejecuta un agregador con los datos locales de cada partición. Como hemos particionado por localización, y como tenemos 3 particiones y sólo 4 localizaciones en nuestro conjunto de datos, el particionado está asignando “France” y “Spain” a una partición, “USA” a otra y “UK” a otra.

Todo ésto es un poco complicado de enteder a primera vista, pero este tipo de experimentos son los que hacen que de verdad dominemos y entendamos una herramienta, así que hay que tener paciencia. Lo que veremos a continuación es bastante más intuitivo.

groupBy

El siguiente código nos parecerá bastante intuitivo:

topology.newStream("spout", spout)
  .groupBy(new Fields("location"))
  .aggregate(new Fields("location"), new Count(), new Fields("count"))
  .each(new Fields("location", "count"), new Utils.PrintFilter());

Pero, ¿Qué hace ésto, en realidad?

...
[France, 25]
[UK, 2]
[USA, 25]
[Spain, 44]
[France, 26]
[UK, 3]
...

Este código nos produce cuentas para cada localización independiente, sin haber especificado paralelismo de ningún tipo. De hecho, estamos usando un agregador muy sencillo (Count()), que simplemente cuenta el número de tuplas. groupBy() lo que hace es crear un flujo de datos agrupado lógicamente por algunos campos. Esta agrupación lógica modifica el comportamiento de las operaciones aggregate(). En vez de agregar todas las tuplas de un batch, que es lo que conocíamos, ahora se nos agrega sólo las tuplas que pertenecen a cada uno de los grupos lógicos, de forma independiente. Así que es como si tuviésemos tantos flujos como grupos diferentes hay, si es que pensarlo así ayuda.

Tengamos en cuenta, por tanto, que groupBy() en sí mismo no es una operación de reparticionado. groupBy() seguido de aggregate() sí lo es, pero groupBy() seguido de partitionAggregate() no. Como deberes, pensad en ello y experimentad con todo ésto.

Conclusión

Hasta aquí hemos visto unas cuantas primitivas fundamentales de Trident. Hay bastantes más cosas de las que no hemos hablado todavía, como la API de estado. Sin embargo, esperamos que los conceptos se hayan introducido de forma clara y eficiente.

Podéis jugar con el proyecto de github e implementar varios ejemplos como:

• Cuentas por cada hashtag.
• Los últimos tres tweets para cada actor.
• Las palabras más usadas para cada actor.
• Las palabras más usadas, en general.
• Los hashtags populares en una cierta ventana temporal.

Algunos de éstos ejemplos involucran guardar estado de algún tipo. La API de estado de Trident es una opción para ello (quizás hablamos de ésto en otro post). Otra opción sería simplemente conectar con una base de datos en un agregador o función. Y otra opción sería mantener estado en memoria en los mismos procesos, pero teniendo en cuenta que ésta última opción no es verdaderamente tolerante a fallos y que no la aconsejamos para ninguna aplicación seria.

Entrada original en: Un vistazo a la API de Trident (Storm)

Splout Cloud está basado en Splout SQL, una base de datos SQL libre para Hadoop.

Motivación

El Big Data está presente en mucho sitios, pero sobretodo lo encontramos en forma de grandes ficheros de datos que se recolectan y se guardan en cualquier lugar. Es un gran reto poder hacer toda ésta información consultable, y sobretodo que las consultas se ejecuten realmente rápido. La solución típica conlleva volcar todos los datos a una base de datos. Pero hay muchos tipos de bases de datos, ¿Cuál deberíamos elegir? ¿Quién se va a ocupar del mantenimiento de la base de datos? ¿Y cómo serán los procesos de ETL que muevan los datos y los actualicen?

¿No sería maravilloso que nuestros ficheros simplemente se volvieran consultables? Splout Cloud te permite dejar todos tus datos en un bucket S3 y éstos datos son sincronizados periódicamente con un sistema escalable de consultas basado en Splout SQL.

Cómo funciona

Así es como Splout Cloud funciona:

• El usuario sincroniza sus datos con un bucket S3. Por ejemplo, subiendo ficheros nuevos cada día.
• Un clúster Splout Cloud se instancia dentro de la cuenta AWS del usuario. Dependiendo de la aplicación y de las necesidades, se ajusta el tamaño del clúster.
• Se define una política de particionado. Por ejemplo, particionar las ventas de los clientes por ID de cliente.
• Splout Cloud sincroniza transparentemente el bucket S3 periódicamente. El usuario puede consultar el clúster a través de una interfaz REST y construir sus propias aplicaciones por encima.

Splout Cloud no es adecuado para actualizaciones de los datos en tiempo real (cada pocos segundos). Splout Cloud se orienta a aplicaciones con grandes volúmenes de datos donde actualizar los datos unas cuantas veces al día es suficiente.

¡Aplica para la beta!

Animamos a cualquiera a que pruebe Splout Cloud con su caso de uso. Para eso puedes aplicar para la beta. Si tu conjunto de datos está por debajo de los 100GB podrás probar el servicio de forma gratuita.

Entrada original en: Presentando Splout Cloud: Un servicio de consultas SQL de baja latencia en la nube.

En este post presentaremos un caso de uso Big Data para una gran tienda de alimentos. Haremos un sumario de las compras individuales de cada cliente usando Apache Pig y volcaremos éste análisis a Splout SQL para poder ser capaces de lanzar consultas en tiempo real sobre él. Así, podremos combinar la información histórica de cada cliente con una lista de productos a promocionar para poder sugerir descuentos personalizados. Esta información se podría usar fácilmente por un sistema de impresión de cupones para fidelización de los clientes.

Combinar una herramienta de desarollo ágil Big Data como Pig con un sistema de consultas flexible y de baja latencia como Splout SQL nos provée una solución simple y efectiva a éste problema, que podremos simular a través de este post apenas sin esfuerzo.

Requerimientos

Para poder seguir las instrucciones de éste post necesitamos:

• Tener Hadoop y Apache Pig instalados. Hemos probado éste ejemplo con Hadoop CDH3 y Pig 0.9.2.
• Tener Python instalado. Los scripts de generación de datos están escritos en Python.
• Descargar el JAR pangool-core versión 0.6.2 o superior. Este JAR puede ser descargado desde Maven central. Lo usaremos en Pig para llamar a una “StoreFunc” propia.
• Tener Splout SQL versión 0.2.2 o superior. Para usar Splout SQL sólo es necesario descargarse la última release, descomprimirla y arrancar los servicios QNode y DNode. Se puede encontrar más información en “Getting started“, en la página oficial.

Preparando los datos de entrada

Habrá dos ficheros como datos de entrada:

• Una lista de productos a promocionar. Hay que ofrecer descuentos de éstos productos a los clientes.
• La base de datos histórica de los pagos de los clientes en la tienda. Ésta es la parte “Big Data” del sistema. Para un “retail” grande, ésta base de datos puede ser enorme. Nosotros generaremos sólo unos cuantos pagos ejemplo.

Para generar los datos de entrada usaremos la herramienta retail_data_generator que hemos creado para éste post. Este archivo contiene dos scripts python y un fichero JSON. El fichero JSON describe las categorías y sub-categorías que se venden en la tienda. Los scripts python generan cada uno de los ficheros de entrada. Tras descomprimir el archivo, podemos ejecutar los siguientes comandos:

python gen_prod_list.py > promotional_products.txt
python gen_tickets.py > tickets_sample.txt

(Se pueden cambiar los parámetros de generación de cada uno de los scripts modificando las variables internas. Los valores por defecto deberían ser suficientemente ilustrativos.)

Podemos ahora subir los ficheros generados al HDFS de Hadoop:

hadoop fs -put promotional_products.txt promotional_products.txt
hadoop fs -put tickets_sample.txt tickets_sample.txt

Sumarización de las compras de los clientes en Pig

Como ejemplo sencillo, calcularemos el sumario de todas las compras de cada cliente para poder obtener las categorías de alimentos que más compra cada uno. Obtendremos el “top categorías” y el “top sub-categorías”, ya que ambas dimensiones nos permitirán entender mejor el comportamiento de cada cliente. El código en Pig Latin para el sumario es el siguiente:

a = LOAD 'tickets_sample.txt' USING PigStorage('\t') AS (date, client_id, category, sub_category);
b = GROUP a BY (client_id, sub_category);
c = FOREACH b GENERATE FLATTEN(group) AS (client_id, sub_category), COUNT(a) AS count;
d = FOREACH c GENERATE (chararray)client_id, (chararray)sub_category, count;
top_subcategories = ORDER d BY client_id, count DESC;

e = GROUP a BY (client_id, category);
f = FOREACH e GENERATE FLATTEN(group) AS (client_id, category), COUNT(a) AS count;
g = FOREACH f GENERATE (chararray)client_id, (chararray)category, count;
top_categories = ORDER g BY client_id, count DESC;

Podemos ahora usar el comando útil “illustrate” de Pig para ver si lo que hemos introducido es realmente lo que queremos o no. El comando ILLUSTRATE muestrea el conjunto de datos de entrada y aplica el “flow” rápidamente sobre los datos muestreados:

ILLUSTRATE top_categories;
ILLUSTRATE top_subcategories;

Finalmente, para ejecutar el “flow” de verdad, podemos usar el comando STORE, y para eso tenemos que cargar antes el JAR pangool-core (sustituyendo la ruta como sea necesario):

REGISTER /home/.../pangool/pangool-core-0.60.2.jar;

Los comandos STORE:

STORE top_subcategories INTO 'retail_top_subcategories' USING com.datasalt.pangool.pig.PangoolStoreFunc('retail_top_subcategories', 'client_id', 'sub_category', 'count');
STORE top_categories INTO 'retail_top_categories' USING com.datasalt.pangool.pig.PangoolStoreFunc('retail_top_categories', 'client_id', 'category', 'count');

Hemos usado una función Pangool propia llamada “PangoolStoreFunc”. Esta función nos permite guardar el output de un proceso Pig en un fichero binario que tiene un esquema definido. Este fichero puede ser entonces procesado fácilmente por Splout SQL sin necesidad de redefinir su schema. El primer argumento de esta función es el nombre de la tabla, y los siguientes argumentos son el nombre de cada columna. Las tablas SQL de Splout se llamarán igual.

Desplegando el análisis a Splout SQL

Ahora que ya hemos ejecutado el sumario de cada cliente, podemos desplegar las tablas resultantes junto con los productos a promocionar a Splout SQL. De esta manera, podremos ejecutar queries SQL sobre las tres tablas en tiempo real. Para ello crearemos un descriptor JSON en el directorio donde hemos instalado Splout SQL llamado “retail_tablespace.json”:

{
        "name": "retail_example",
        "nPartitions": 2,
        "partitionedTables": [{
                "name": "retail_top_categories",
                "partitionFields": "client_id",
                "tableInputs": [{
                        "inputType": "TUPLE",
                        "paths": [ "retail_top_categories" ]
                }]
         },{
                "name": "retail_top_subcategories",
                "partitionFields": "client_id",
                "tableInputs": [{
                        "inputType": "TUPLE",
                        "paths": [ "retail_top_subcategories" ]
                }]
        }],
        "replicateAllTables": [{
                "name": "promotional_products",
                "schema": "category:string, sub_category:string, product:string",
                "tableInputs": [{
                        "paths": [ "promotional_products.txt" ]
                }]
        }]
}

Hemos especificado dos tablas a ser particionadas (retail_top_categories, retail_top_subcategories) y una para ser replicada en cada partición (promotional_products). Las primeras dos tablas vienen de Pig y son potencialmente Big Data. Splout SQL usa particionamiento de datos para poder escalar, así que necesitamos definir una estrategia de particionamiento. Como sólo estamos interesados en consultas concretas sobre cada cliente, podemos particionar fácilmente por “client_id”. Aquí hemos fijado dos particiones, pero para poder escalar en un clúster real tendríamos que incrementar este número. Por otro lado, la tercera tabla es un fichero relativamente pequeño que puede replicarse en cada partición, y así podemos ejecutar joins para cada cliente con ella.

Las tablas que exportamos de Pig se importan con el tipo “TUPLE”, dado que las hemos guardado en formato binario Pangool. Para la tercera tabla hemos omitido algunos valores por defecto. El tipo por defecto es “TEXT” y se importa usando tabulación como separador, que es exactamente lo que queríamos para este caso.

Ahora podemos invocar los comandos apropiados de generación y despliegue en Splout SQL:

hadoop jar splout-*-hadoop.jar generate -tf file:///`pwd`/retail_tablespace.json -o out-retail-example
hadoop jar splout-hadoop-*-hadoop.jar deploy -root out-retail-example -ts retail_example -q http://localhost:4412

El primer comando crea los ficheros binarios SQLite que después serán desplegados a Splout SQL a través del segundo comando.

Implementando el sistema de generación de cupones

Para ofrecer cupones a los clientes podemos ejecutar un join como el siguiente:

SELECT product FROM promotional_products, retail_top_categories WHERE client_id = 0 AND retail_top_categories.category = promotional_products.category ORDER BY count DESC LIMIT 5;

Esta query seleccionará 5 productos de la lista de productos promocionales basados en las compras históricas del cliente. El join está basado en la categoría del producto, pero podríamos haber usado también la sub-categoría:

SELECT product FROM promotional_products, retail_top_subcategories WHERE client_id = 0 AND retail_top_subcategories.sub_category = promotional_products.sub_category ORDER BY count DESC LIMIT 5;

Generalmente, ambas queries devolverán resultados diferentes. Ver el consumo histórico de un cliente basado en la categoría de los alimentos que consume o en la sub-categoría nos ofrece diferentes puntos de vista. Podemos probar ambas queries SQL en la webapp de administración de Splout (en localhost:4412).

En el siguiente HTML adjunto hemos implementado una herramienta ilustrativa simple Javascript que imprime cupones haciendo queries a Splout SQL.

Conclusión

Combinando Pig con Splout SQL hemos podido implementar un sistema simple de generación de cupones casi sin esfuerzo. Este sistema puede escalar a grandes cantidades de datos de entrada. Para muchísimo clientes comprando productos durante largos periodos de tiempo podemos analizar nuestros datos de forma escalable usando Hadoop (Pig) y además ser capaces de consultar este análisis en tiempo real con Splout SQL. El ejemplo que hemos implementado es muy sencillo, pero no es difícil imaginar variantes más complejas como la implementación de aplicaciones móviles para los clientes que les permitan consultar el consumo histórico y la evolución en el tiempo de su consumo.

Entrada original en: Pig + Splout SQL para un sistema de cupones promocionales: Un romance Big Data

En éste post presentaremos un ejemplo de aplicación Big Data para analizar tweets y mostrar informes consolidados, estadísticas relevantes a usuarios de Twitter a través de una webapp interactiva de baja latencia. Para eso juntaremos Hive (una solución open-source de “warehousing” que permite procesos analíticos fáciles sobre Hadoop) con Splout SQL, una base de datos SQL particionada de baja latencia para Hadoop. Construiremos un simple aunque potente y escalable panel tipo el de “Tweet archivist” y lo haremos sin escribir ni una sola línea de código. La herramienta proveerá menciones históricas, hashtags y tweets populares para cada actor en el dataset de entrada analizado.

Los datos de Twitter

Construir herramientas que sean capaces de analizar tal cantidad de datos es todo un reto. En este post enseñaremos cómo usar una herramienta de procesamiento escalable (Hive) y una base de datos escalable de baja latencia (Splout SQL), y veremos lo bien que se integran la una con la otra.

Requerimientos

Para poder seguir de forma práctica este post necesitaremos:

• Tener Hadoop y Hive instalados. Lo hemos probado con Hadoop CDH3 y Hive 0.10.0.
• Tener “hive-site.xml” definido y el directorio conf/ de Hive añadido al CLASSPATH / HADOOP_CLASSPATH. “hive-default.xml” está deprecated y puede no funcionar bien.
• Tener Splout SQL. Para usar Splout SQL sólo tenemos que descargar la última distribución, descomprimirla y arrancar los servicios QNode y DNode. Se puede encontrar más información al respecto en “Getting started” en la web oficial.

Preparando el conjunto de datos de entrada (tweets)

Podemos descargar algunos tweets de ejemplo de la API REST de Twitter usando “curl“. Usaremos algunas cuentas populares como “BBCNews” o “Reuters”. Tenemos que convertir el array de tweets que Twitter nos devuelve en un tweet independiente por línea, y para eso usaremos “sed“:

mkdir temp
cd temp
curl "http://api.twitter.com/1/statuses/user_timeline.json?screen_name=TechCrunch&count=200" | sed "s/},{/}\n{/g" > techcrunch_tweets.json
curl "http://api.twitter.com/1/statuses/user_timeline.json?screen_name=reuters&count=200" | sed "s/},{/}\n{/g" > reuters_tweets.json
curl "http://api.twitter.com/1/statuses/user_timeline.json?screen_name=BBCNews&count=200" | sed "s/},{/}\n{/g" > bbcnews_tweets.json
curl "http://api.twitter.com/1/statuses/user_timeline.json?screen_name=barackobama&count=200" | sed "s/},{/}\n{/g" > barackobama_tweets.json
curl "http://api.twitter.com/1/statuses/user_timeline.json?screen_name=ladygaga&count=200" | sed "s/},{/}\n{/g" > ladygaga_tweets.json

Vamos a usar Hadoop con Hive, así que necesitamos tener ambas cosas instaladas. Una vez tenemos Hadoop funcionando, podemos crear un directorio en el HDFS y subir todos los ficheros JSON que nos hemos descargado:

hadoop fs -mkdir tweets
hadoop fs -put *.json tweets/

Finalmente, abrimos Hive y creamos una tabla externa apuntando a éste “contenedor de tweets” (con cuidado de sustituir “myuser” por el usuario que sea):

  CREATE EXTERNAL TABLE tweets
    (jsonData STRING)
  ROW FORMAT DELIMITED FIELDS
    TERMINATED BY '\n'
    LOCATION '/user/myuser/tweets';

JSON + Hive

ADD JAR %HIVE_HOME%/lib/hive-contrib-%HIVE_VERSION%.jar;

La siguiente query parseará el campo “text” en cada JSON::

  SELECT
    a.text
  FROM tweets
    LATERAL VIEW json_tuple(tweets.jsonData, 'text') a AS text;

La siguiente query extrae la primera mención de cada tweet que tenga menciones:

  SELECT
    regexp_extract(a.text, "(@[^ ]*)", 1)
  FROM tweets
    LATERAL VIEW json_tuple(tweets.jsonData, 'text') a AS text;

Creando vistas con Hive

Crearemos tres vistas: la de menciones, la de retweets y la de hashtags. En la de menciones tendremos toda interacción entre dos usuarios de Twitter y la fecha en que ocurrió esa interacción. En la de retweets tendremos todos los tweets populares (“retweet_count” > 0) y en la de hashtags el número de veces que cada usuario usa un cierto hashtag. Crear todas estás vistas es tan fácil como ejecutar una query Hive para cada una de ellas.

La vista de menciones puede ser creada con la siguiente query:

  CREATE TABLE mentions AS
    SELECT
      foo.screenname as mentioner,
      substr(wordTable.word, 2) as mentioned,
      to_date(from_unixtime(unix_timestamp(substr(foo.ttime, 5), 'MMM dd hh:mm:ss ZZZZZ yyyy'))) as daydate
    FROM (
      SELECT
        c.screenname,
        a.ttime,
        split(a.text, '[^a-zA-Z_0-9@#]') AS wordarray
      FROM tweets
        LATERAL VIEW json_tuple(jsonData, 'created_at', 'text') a AS ttime, text
        LATERAL VIEW json_tuple(jsonData, 'user') b AS user
        LATERAL VIEW json_tuple(b.user, 'screen_name') c AS screenname
      ) foo
      LATERAL VIEW explode(foo.wordarray) wordTable AS word
    WHERE wordTable.word LIKE '@%';

Si hacemos una query tipo SELECT … LIMIT 10 veremos cómo es la tabla creada:

Tenemos un “mentioner” que mencionó a alguien (“mentioned”) y la fecha en que la mención sucedió.

La tabla de retweets es muy simple de generar y puede ser creada con la siguiente query:

CREATE TABLE retweets AS
  SELECT
    c.screenname as user,
    a.text as tweet,
    to_date(from_unixtime(unix_timestamp(substr(a.time, 5), 'MMM dd hh:mm:ss ZZZZZ yyyy'))) as daydate,
    a.retweetcount as retweetcount
  FROM tweets
    LATERAL VIEW json_tuple(jsonData, 'created_at', 'text', 'retweet_count') a AS time, text, retweetcount
    LATERAL VIEW json_tuple(jsonData, 'user') b AS user
    LATERAL VIEW json_tuple(b.user, 'screen_name') c AS screenname
  WHERE a.retweetcount > 0;

Esta tabla tiene un usuario (“user”, el autor del tweet), el texto del tweet (“text”), la fecha en fue posteado y el número de veces que fue retwiteado:

Por último, la tabla de hashtags:

CREATE TABLE tophashtags AS
  SELECT
    foo2.screenname AS user,
    foo2.word AS word,
    COUNT(*) as count
  FROM (
    SELECT
      foo.screenname as screenname,
      wordTable.word as word FROM (
        SELECT
          c.screenname,
          a.time,
          split(a.text, "[^a-zA-Z_0-9@#]+") AS wordarray
        FROM tweets
          LATERAL VIEW json_tuple(jsonData, 'created_at', 'text') a AS time, text
          LATERAL VIEW json_tuple(jsonData, 'user') b AS user
          LATERAL VIEW json_tuple(b.user, 'screen_name') c AS screenname
      ) foo
      LATERAL VIEW explode(foo.wordarray) wordTable AS word
    WHERE wordTable.word LIKE '#%'
  ) foo2
  GROUP BY foo2.screenname, foo2.word;

Ésta tabla simplemente tiene el usuario, el hashtag usado y el número de veces que éste usuario usó éste hashtag.

Volcando vistas Hive a Splout SQL

De nada sirve tener toda esta información generada si no podemos hacer queries rápidas contra ella desde una aplicación web. Queremos poder renderizar un panel para cada usuario donde éste navegará por todas sus estadísticas, y queremos hacer ésto de manera escalable, óptima y flexible. En vez de precomputar todo el panel para volcarlo a una base de datos llave / valor, simplemente volcaremos las tablas generadas en Hive a Splout SQL, que será la que alimentará el frontend.

{
	"name": "hive_splout_example",
	"nPartitions": 2,
	"partitionedTables": [{
		"name": "retweets",
		"partitionFields": "user",
	    	"tableInputs": [{
			"inputType": "HIVE",
			"hiveTableName": "retweets",
			"hiveDbName": "default"
		}]
	},{
		"name": "mentions_of_me",
		"partitionFields": "mentioned",
	    	"tableInputs": [{
			"inputType": "HIVE",
			"hiveTableName": "mentions",
			"hiveDbName": "default"
		}]
	},{
		"name": "mentions_by_me",
		"partitionFields": "mentioner",
	    	"tableInputs": [{
			"inputType": "HIVE",
			"hiveTableName": "mentions",
			"hiveDbName": "default"
		}]
	},{
		"name": "tophashtags",
		"partitionFields": "user",
	    	"tableInputs": [{
			"inputType": "HIVE",
			"hiveTableName": "tophashtags",
			"hiveDbName": "default"
		}]
	}]
}

En éste fichero JSON estamos describiendo el “tablespace” que desplegaremos en Splout SQL, que estará compuesto de cuatro tablas. Fijaros que particionamos por el usuario, y que importamos la misma tabla “mentions” dos veces. Ésto es porque estamos interesados tanto en las menciones de un usuario a otro como en las menciones que un usuario recibe de otros. Dado que debemos particionar el conjunto de datos, tenemos pues que importar doblemente la tabla, primero particionada por el que menciona y luego por el que es mencionado.

Para desplegar el “tablespace” a Splout SQL podemos usar la siguiente línea de comandos, ejecutada desde el directorio de instalación de Splout SQL:

hadoop jar splout-*-hadoop.jar generate -tf file:///`pwd`/hive_splout_example.json -o out-hive_splout_example
hadoop jar splout-hadoop-*-hadoop.jar deploy -q http://localhost:4412 -root out-hive_splout_example -ts hive_splout_example

Obteniendo los resultados finales

Podemos ir al panel de administración de Splout SQL para ver que el “tablespace” se ha desplegado correctamente, y ejecutar algunas “queries” de prueba. Por ejemplo, podemos obtener los hashtags más usado por “BarackObama” como ilustra la siguiente imagen:

Y para usar ésta información en una webapp podemos usar la API REST de Splout SQL: http://localhost:4412/api/query/hive_splout_example?sql=SELECT * FROM tophashtags WHERE user = ‘BarackObama’ ORDER BY count DESC;&key=BarackObama

La cual devolverá una respuesta JSON:

{
  millis: 10,
  error: null,
  result: [
  {
    count: 29,
    word: "#WeDemandAVote",
    user: "BarackObama"
  },
  {
    count: 28,
    ...
  }]
}

La siguiente tabla muestra las queries que podemos hacer a Splout SQL para poder mostrar un panel de información para cada usuario:

Pero también podríamos hacer cosas más complejas como gráficas de evolución temporal de menciones, ya que tenemos la fecha en cada mención. Así como agregados parciales por periodos temporales (mes, semana, año, …), y un largo etcétera.

Conclusión

Hemos podido construir una aplicación de analítica sobre Twitter totalmente escalable, ¡Y lo hemos hecho sin escribir una sola línea de código! Ésto ha sido posible por la integración entre Hive y Splout SQL. Hive provée una forma fácil e interactiva de generar “vistas Big Data” con datos en Hadoop, mientras que Splout SQL puede importar esas vistas directamente e indexarlas para que los usuarios puedan hacer queries rápidas sobre ellas. Teniendo SQL tanto en el backend como en el frontend podemos ser muy flexibles y ágiles. ¡No dudes en probar éste ejemplo y darnos tu feedback!

Entrada original en: Hive + Splout SQL para analítica sobre Twitter: Un romance Big Data

Cascading

Cascading es un framework que permite a programadores Java construir fácilmente flujos complejos Hadoop, abstrayéndoles de Map/Reduce. Nos gusta Cascading porque nos permite prototipar rápidamente. Hay otras herramientas que no son Java por encima de Cascading que se usan comúnmente como Cascalog o Scalding. A través de este ejemplo veremos unas cuantas primitivas Cascading y cómo se interconectan entre ellas.

Splout SQL

Ya hemos presentado Splout SQL y visto un ejemplo de arquitectura lambda con él. Splout SQL permite servir Big Data a bajas latencias y alta carga como cualquier otra NoSQL – pero con la obvia diferencia de que es SQL. A diferencia de los sistemas tipo Dremel que sirven para hacer queries exploratorias ad-hoc, análisis offline, etc, Splout SQL está pensado para ser usado en una web, ya que proveé un rendimiento predecible y consistente ante escenarios concurrentes. Y además, como veremos en éste ejemplo, permite desplegar datos desde Hadoop muy fácilmente.

La webapp “back office” de un portal e-Commerce

Éste imaginario portal e-Commerce tiene muchos usuarios y muchas categorías de productos. La categoría está siempre presente en la URL de cada producto. Hay un departmento de atención al usuario que se encarga de atender posibles problemas, contestar llamadas y planear campañas de fidelización. Necesitamos construir el backend de una aplicación web que se usará en este departmento y cuyos requerimientos son:

• Para cualquier usuario, obtener la secuencia de eventos que éste ha hecho en la web, en cualquier intervalo de tiempo. (Ésto nos ayudará a detectar y analizar posibles problemas, y es una información muy buena para el departamento técnico, para poder arreglar bugs fácilmente).

• Para cualquier usuario, poder visualizar un resumen de su actividad en las diferentes categorías, y así poder llevar a cabo campañas de fidelización eficaces. (Por ejemplo, conocer las cinco categorías en las que el usuario interactúa más nos puede permitir ofrecer descuentos u otros productos promocionales interesantes para cada usuario).

• Escalable tanto en el procesamiento como en el servicio. La cantidad de datos a ser servidos es igual de grande que la cantidad de datos a ser analizados.
• Sencilla de implementar.
• Flexible – para que podamos añadir y cambiar estadísticas fácilmente, cambiar toda la lógica de negocio y poder recalcularlo todo sin problema.
• Con un coste razonable.

La solución

En la solución – que se puede ver en github – los logs Apache son parseados y analizados con Cascading, que produce dos ficheros de salida: uno con los logs parseados tal cual y otro con estadísticas consolidadas a partir de ellos (un agrupado por usuario, categoría y fecha). Ambos ficheros de salida pueden ser entonces transformados en tablas SQL para Splout SQL y consultados en tiempo real desde la aplicación web.

Usar Cascading para el procesamiento nos permite desarrollar e iterar rápidamente. Usar Splout SQL para el servicio nos permite ejecutar queries flexibles SQL sobre los datasets analizados y escalar horizontalmente sin tener que tener un sistema complejo y caro por debajo.

Explicaremos cada parte de la solución a continuación.

Procesamiento de logs

En primer lugar, los logs que se producen en el portal han de ser guardados en el HDFS para que los procese Hadoop. En ésta parte no vamos a incidir mucho – pero si la solución sencilla de subir periódicamente ficheros de logs parciales al HDFS no fuera suficiente, hay herramientas que pueden ayudarnos como Flume, Scribe, Kafka o Storm.

Una vez en el HDFS, los ficheros pueden ser analizados usando Cascading muy fácilmente. Sólo tenemos que parsearlos y emitirlos, junto con una agregación diaria por usuario y categoría. Éste es todo el código Java que necesitamos para implementar la lógica de negocio:

	public void indexLogs(String inputPath, String outputPathLogs, String outputPathAnalytics) {
		// define what the input file looks like, "offset" is bytes from beginning
		TextLine scheme = new TextLine(new Fields("offset", "line"));

		// create SOURCE tap to read a resource from the local file system, if input is not an URL
		Tap logTap = inputPath.matches("^[^:]+://.*") ? new Hfs(scheme, inputPath) : new Lfs(scheme,
		    inputPath);

		// declare the field names we will parse out of the log file
		Fields apacheFields = new Fields("ip", "user", "time", "method", "category", "page", "code", "size");

		// define the regular expression to parse the log file with
		String apacheRegex = "^([^ ]*) +[^ ]* +([^ ]*) +\\[([^]]*)\\] +\\\"([^ ]*) /([^/]*)/([^ ]*) [^ ]*\\\" ([^ ]*) ([^ ]*).*$";

		// declare the groups from the above regex we want to keep. each regex group will be given
		// a field name from 'apacheFields', above, respectively
		int[] allGroups = { 1, 2, 3, 4, 5, 6, 7, 8 };

		// create the parser
		RegexParser parser = new RegexParser(apacheFields, apacheRegex, allGroups);

		// create the input analysis Pipe
		Pipe parsePipe = new Each("logs", new Fields("line"), parser, Fields.RESULTS);

		// parse the date and split it into day + month + year
		parsePipe = new Each(parsePipe, new Fields("time"), new DateParser(
		    new Fields("day", "month", "year"), new int[] { Calendar.DAY_OF_MONTH, Calendar.MONTH,
		        Calendar.YEAR }, "dd/MMM/yyyy:HH:mm:ss"), Fields.ALL);

        // aggregate by date, user and category
		Pipe analyzePipe = new GroupBy("analyze", parsePipe, new Fields("day", "month", "year", "user",
		    "category"));
        // count each aggregation
		analyzePipe = new Every(analyzePipe, new Count());

		// create a SINK tap to write to the default filesystem
		// To use the output in Splout, save it in binary (SequenceFile).
		// In this way integration is both efficient and easy (no need to re-parse the file again).
		Tap remoteLogTap = new Hfs(new SequenceFile(Fields.ALL), outputPathLogs, SinkMode.REPLACE);
		Tap remoteAnalyticsTap = new Hfs(new SequenceFile(Fields.ALL), outputPathAnalytics, SinkMode.REPLACE);

		// set the current job jar
		Properties properties = new Properties();
		AppProps.setApplicationJarClass(properties, LogIndexer.class);

		Map<String, Tap> sinks = new HashMap<String, Tap>();
		sinks.put("logs", remoteLogTap);
		sinks.put("analyze", remoteAnalyticsTap);

		// connect the assembly to the SOURCE and SINK taps
		Flow parsedLogFlow = new HadoopFlowConnector(properties).connect(logTap, sinks, parsePipe,
		    analyzePipe);

		// start execution of the flow (either locally or on a cluster)
		parsedLogFlow.start();

		// block until the flow completes
		parsedLogFlow.complete();
	}

La variable “parsePipe” es la responsable de parsear los logs, y “analyzePipe” ejecuta la agregación sobre ellos. Usamos ambas variables como outputs del flow (sinks). Vemos que parsear los logs es muy fácil usando la función RegexParser.

Fijaros en lo fácil que es razonar sobre el flujo de datos con las operaciones Each, GroupBy y Every. Ésto es similar a lo que vimos con Trident en éste post previo. De alguna manera, Trident es al procesamiento tiempo-real lo que Cascading es al procesamiento por lotes.

Indexado y servicio de los logs

Para poder servir eficientemente toda la información estructurada que hemos sacado de Cascading necesitamos una base de datos escalable, rápida y consistente. Usando Splout SQL podemos también acceder a la expresividad SQL sobre los datos de manera que no tenemos que precomputarlo todo con antelación. Podemos implementar gráficas de evolución temporal tipo las de Google Analytics sin apenas coste.

Por otro lado, para poder usar Splout SQL debemos particionar los datos. Pero en nuestro caso ésto no es un problema ya que queremos proveer un panel individualizado para cada usuario, así que podemos particionar fácilmente por usuario.

	public void deployToSplout(String outputPathLogs, String outputPathAnalytics, String qNode,
	    int nPartitions) throws Exception {
		// add sqlite native libs to DistributedCache
		if(!FileSystem.getLocal(conf).equals(FileSystem.get(conf))) {
			SploutHadoopConfiguration.addSQLite4JavaNativeLibsToDC(conf);
		}

		// delete tablespace-generated files if they already exist
		FileSystem outputPathFileSystem = new Path(outputPath).getFileSystem(conf);
		Path outputToGenerator = new Path(outputPath + "-generated");
		if(outputPathFileSystem.exists(outputToGenerator)) {
			outputPathFileSystem.delete(outputToGenerator, true);
		}

		TablespaceBuilder builder = new TablespaceBuilder();
		// build a Table instance of each table using the builder
		TableBuilder logsTable = new TableBuilder("logs", getConf());
		String[] logsColumns = new String[] { "ip", "user", "time", "method", "category", "page", "code", "size", "day", "month", "year" };
		logsTable.addCascadingTable(new Path(outputPathLogs), logsColumns, conf);
		logsTable.partitionBy("user");

		TableBuilder analyticsTable = new TableBuilder("analytics", getConf());
		String[] analyticsColumns = new String[] { "day", "month", "year", "user", "category", "count" };
		analyticsTable.addCascadingTable(new Path(outputPathAnalytics), analyticsColumns, conf);
		analyticsTable.partitionBy("user");

		builder.add(logsTable.build());
		builder.add(analyticsTable.build());
		// define number of partitions
		builder.setNPartitions(nPartitions);

		// instantiate and call the TablespaceGenerator with the output fo the TablespaceBuilder
		TablespaceGenerator viewGenerator = new TablespaceGenerator(builder.build(), outputToGenerator,
		    this.getClass());
		viewGenerator.generateView(conf, SamplingType.DEFAULT, new DefaultSamplingOptions());

		// finally, deploy the generated files
		StoreDeployerTool deployer = new StoreDeployerTool(qNode, conf);
		List specs = new ArrayList();
		specs.add(new TablespaceDepSpec("cascading_splout_logs_example", outputToGenerator.toString(), 1, null));

		deployer.deploy(specs);
	}

Fijaros cómo usamos el método addCascadingTable en TableBuilder para poder indexar ficheros binarios Cascading directamente. Ésta manera es la más eficiente de exportar la salida de un proceso Cascading ya que así nos ahorramos tener que exportarlo a texto y reparsearlo después.

En éste ejemplo hemos usado la API Java para exportar a Splout SQL, pero bien podríamos haberlo hecho con línea de comandos sin programar nada. Para hacer lo mismo que en el código de arriba sin programar, podríamos ejecutar los siguientes comandos:

  hadoop jar splout-*-hadoop.jar generate -tf file:///`pwd`/cascading-export.json -o out-cascading-export
  hadoop jar splout-*-hadoop.jar deploy -root out-cascading-export -ts cascading_splout_logs_example -q http://localhost:4412

Teniendo en cuenta que el fichero “cascading-export.json” sería algo como:

{
	"name": "cascading_splout_logs_example",
	"nPartitions": 2,
	"partitionedTables": [{
		"name": "logs",
		"partitionFields": "user",
		"tableInputs": [{
			"inputType": "CASCADING",
			"cascadingColumns": "ip,user,time,method,category,page,code,size,day,month,year",
			"paths": [ "out-clogs-logs" ]
		}]
	},{
		"name": "analytics",
		"partitionFields": "user",
		"tableInputs": [{
			"inputType": "CASCADING",
			"cascadingColumns": "day,month,year,user,category,count",
			"paths": [ "out-clogs-analytics" ]
		}]
	}]
}

Pruébalo!

En ésta aplicación web para un portal e-Commerce hemos ilustrado la integración fácil entre Splout SQL y herramientas Hadoop como Cascading. En los siguientes posts haremos lo mismo para Hive y Pig. El proyecto y las instrucciones detalladas para probarlo están en github, así que prúebalo!

Entrada original en: Cascading + Splout SQL para analizar y servir logs: Un romance Big Data

Así que para cualquier hashtag queremos ser capaces de hacer una query a un servicio remoto y obtener una evolución temporal parecida a:

{
  "20091022":115,
  "20091023":115,
  "20091024":158,
  "20091025":19
}

Arquitectura lambda

La “arquitectura lambda” es un concepto acuñado por Nathan Marz. A nosotros nos gusta porque es un modelo que escala y que tiene tanto las ventajas del procesamiento en lotes (batch) como la frescura del proceso de datos en tiempo real, con actualización de los datos en segundos. No hemos encontrado ejemplos prácticos de éste tipo de arquitectura en Internet así que hemos pensado que sería buena idea desarrollar uno y compartirlo.

Trident

Trident es una API por encima de Storm. Ya hemos hablado antes de Storm y hemos enseñado un ejemplo de uso sencillo en éste post. Trident provée primitivas de alto-nivel por encima de Storm, parecidas a las que provée Cascading para Hadoop (each(), groupBy()). También provée “wrappers” y primitivas para poder guardar o consultar estado en una topología, ya sea en memoria o en una base de datos persistente.

Splout SQL

Splout SQL es una base de datos desarrollada por nosotros que puede desplegar datos de Hadoop de forma muy eficiente. Sería como ElephantDB pero con SQL. Es una base de datos SQL particionada, de sólo lectura y con un alto rendimiento, que puede trabajar con conjuntos de datos generados por Hadoop. Ya la hemos presentado antes en éste post.

Juntando todas las piezas para formar “una lambda”

Los tweets son introducidos en el sistema (1), por ejemplo a través de una cola (Storm tiene conectores para Kafka, JMS y Kestrel, pero también puedes programarte tú mismo uno fácilmente). Un “stream” de Trident (2) guarda los tweets en Hadoop (en el HDFS) y además los procesa en tiempo real para crear un estado en memoria que almacenará las apariciones agrupadas por fecha. En Hadoop (3), donde tendremos todo el histórico de los tweets, podemos ejecutar un proceso “batch” que agregue los tweets por hashtag y fecha y genere un gran fichero como salida. Después de eso, podemos usar Splout SQL para indexar ese fichero y desplegarlo a un clúster Splout SQL (4), el cual será capaz de servir todas las estadísticas de forma altamente eficiente. Entonces, un segundo “stream” de Trident (DRPC) puede ser usado para servir las queries (5), y éste hará una query tanto al resultado de la capa “batch” (a través de Splout SQL) como a la capa de tiempo real (a través del estado en memoria del primer “stream”), y juntar los resultados en una única respuesta. Veremos cada una de éstas cosas en más detalle.

La capa “batch”

La capa “batch” es la más sencilla. Sólo tenemos que almacenar todos los tweets en el HDFS y ejecutar un trabajo periódico que los agregue por hashtag y fecha. No hemos desarrollado esta parte, en vez de ello hemos usado un pequeño conjunto de datos de ejemplo de Infochimps. A continuación mostramos un ejemplo de cómo son éstos datos (que también podeis ver en éste link):

hashtag	2009081313	1	calirfonia
hashtag	2009101713	1	caliroadtrip
hashtag	2009101815	2	caliroadtrip
hashtag	2009080813	1	caliroots
hashtag	2009092807	1	caliroots

Hay un montón de información en Internet sobre cómo hacer tareas sencillas de agregación con herramientas como (de más bajo nivel a menos): Pangool, Cascading, Pig o Hive. La idea es que la salida del proceso de la capa “batch” debería ser algo como lo que hemos mostrado: un fichero tabulado con las cuentas para cada hashtag y fecha. Dado que guardamos todos los tweets en el HDFS, podemos ejecutar un proceso “batch” que calcule muchas cosas, y ejecutarlo tantas veces queramos y recalcularlo todo cada vez. Tenemos absoluta libertad y tolerancia a fallos en ésta capa.

La capa “de servicio”

Uno de los retos de ésta arquitectura ejemplo es ser capaces de servir queries a muy baja latencia y a muy alto rendimiento (con muchos usuarios concurrentes) para un conjunto de datos que puede ser de múltiples GBytes o incluso TBytes. Dado que la mayor parte de los datos vienen de Hadoop (menos la parte de tiempo real), necesitamos algún tipo de puente entre Hadoop y una base de datos rápida. No podemos hacer queries directamente sobre ficheros Hadoop. Cosas como Impala o Drill nos permitirían cierto tipo de queries exploratorias, pero no podemos usarlas delante de un servidor web y esperar servir millones de queries con ellas. Necesitamos índices, B-Trees que hagan las queries verdaderamente eficientes.

Podríamos volcar las estadísticas a una base de datos llave/valor como Voldemort o ElephantDB, donde la llave sería el hashtag. Pero, ¿Qué pasa si queremos ser capaces de proveer estadísticas diarias, semanales y mensuales? ¿Y si queremos mostrar un agrupado por hora para ver en qué horas se menciona más un hashtag? Al usar una base de datos llave/valor nos vemos forzados a precalcularlo todo en Hadoop, lo cual no es siempre muy satisfactorio que digamos. Por eso Splout SQL es más conveniente para este tipo de tareas, ya que nos permite usar SQL para ejecutar queries más flexibles sobre los datos pre-agregados.

En el ejemplo que tenemos en github encontrareis las instrucciones (en inglés) para descargar Splout SQL rápidamente y cargar el conjunto de datos de ejemplo. Los comandos que hay que ejecutar son algo así:

hadoop jar splout-hadoop-*-hadoop.jar simple-generate -i sample-hashtags -o out-hashtags -pby hashtag -p 2 -s "label:string,date:string,count:int,hashtag:string" --index "hashtag,date" -t hashtags -tb hashtags
hadoop jar splout-hadoop-*-hadoop.jar deploy -q http://localhost:4412 -root out-hashtags -ts hashtags

El primer comando genera las estructuras de datos indexadas (ficheros SQL) para poder servir las queries de forma eficiente, y el segundo comando lanza una tarea de despliegue de los datos al clúster Splout SQL.

Como Splout es una SQL particionada, necesitamos especificar un esquema y campos de particionado, así como el número de particiones que queremos generar. Podemos especificar cosas como los índices a generar, ya que sabemos qué tipo de queries vamos a hacer. En este caso hemos creado un índice compuesto por “hashtag” y fecha (“date”) ya que así podríamos extender la aplicación, pudiendo hacer consultas entre periodos arbitrarios de fechas.

Pero para poder usar Splout SQL con Trident necesitamos programar un conector (SploutState). Éste conector extiende “ReadOnlyState” y provée una “StateFactory”, que es lo que nos requiere la API de Trident:

public class SploutState extends ReadOnlyState {
  // The Splout Java client
  private SploutClient sploutClient;
  ...
  public static class Factory implements StateFactory {
    ...
  }
}

Entonces implementa un método que puede recibir múltiples queries y ejecutarlas usando el cliente Java de Splout:

public List<Object> querySplout(String tablespace, List<String> sql, List<String> keys) {
  ...
}

(Fijaros que cada query está asociada con una llave (“keys”). Ésto es necesario porque Splout hace una query a una única partición, y la llave le dice cuál es esa partición.)
(¿Por qué múltiples queries a la vez? Porque Trident agrupará varias tuplas en pequeños lotes que entonces se ejecutarán de golpe de forma más eficiente que una a una.)

Otra cosa que tenemos que hacer es implementar un BaseQueryFunction (HashTagsSploutQuery) que usaremos junto al SploutState para definir la lógica del “stream” DRPC. Esta función contendrá la lógica de negocio involucrada en hacer la query a Splout para éste ejemplo de uso. Así que podríamos reusar SploutState con cualquier otro BaseQueryFunction. En este caso, la función es tal que:

public class HashTagsSploutQuery extends BaseQueryFunction<SploutState, Object> {

  public List<Object> batchRetrieve(SploutState state, List<TridentTuple> args) {
    List<String> sqls = new ArrayList<String>();
    List<String> partitionKeys = new ArrayList<String>();
    // fill the data
    for(TridentTuple arg: args) {
      String hashTag = arg.getString(0);
      sqls.add("SELECT SUM(count), substr(date, 0, 9) as day FROM hashtags WHERE hashtag = '" + hashTag + "' GROUP BY day;");
      partitionKeys.add(hashTag);
    }
    return state.querySplout(TABLESPACE, sqls, partitionKeys);
  }

  public void execute(TridentTuple tuple, Object result, TridentCollector collector) {
    collector.emit(new Values(result));
  }
}

El primer método, batchRetrieve(), es llamado por Trident con un conjunto de tuplas. Entonces llamamos a SploutState para resolver tantas queries SQL como tuplas hemos recibido, particionando por hashtag. El segundo método, execute(), es llamado por Trident para obtener los datos que se añadirán a la tupla del “stream”. En este caso añadiremos un campo nuevo con el resultado de la query.

Más adelante veremos cómo se conecta esta parte con el resto del sistema.

La capa de “tiempo real”

La capa de tiempo real la hemos implementado usando un “stream” Trident que guarda cierto estado en un mapa en memoria. El código puede ser visto en la clase que define la topología (LambdaHashTagsTopology) y es algo así:

TridentState hashTagCounts = topology
  .newStream("spout1", spout)
  .each(new Fields("tweet", "date"), new Split(), new Fields("word"))
  .each(new Fields("word", "date"), new HashTagFilter(), new Fields("hashtag"))
  .groupBy(new Fields("hashtag"))
  .persistentAggregate(mapState, new Fields("hashtag", "date"), new CountByDate(), new Fields("datecount"));

Cuando desarrollamos “streams” Trident tenemos que tener ciertas cosas claras. La primera es la forma en que las tuplas mutan a través del “stream”. Las funciones each() procesan un conjunto de campos de entrada y emiten una tupla con éstos mismos campos junto con los campos de salida que hemos declarado. Por otro lado, las funciones aggregate() sólo emiten los campos que declaramos de salida. Y si queremos emitir un subconjunto de campos tras un each(), podemos usar project(). Por otro lado, y como consecuencia de ésto, no podemos declarar un campo de salida que se llama igual que uno de entrada: colisionaría en la tupla de salida.

Desarrollar “streams” Trident sigue el mismo patrón de programación que desarrollar Cascading. Hay que crear funciones o filtros propios que ejecuten cierta lógica de negocio que luego inyectaremos en la declaración del “stream”. Para este caso, hemos creado “HashTagFilter” que procesa como input una palabra y sólo la emite si es un hashtag (#), en cuyo caso emite todos los caracteres de la palabra menos el primero:

public static class HashTagFilter extends BaseFunction {
  public void execute(TridentTuple tuple, TridentCollector collector) {
    String word = tuple.getString(0);
    if(word.startsWith("#")) {
      collector.emit(new Values(word.substring(1, word.length())));
    }
  }
}

También hemos creado un agregador que se llama “CountByDate” y que se ejecuta antes de persistir las tuplas que agregamos por hashtag. Como queremos mostrar una evolución temporal, necesitamos agrupar las cuentas por fecha. Por simplicidad lo hemos hecho por día, pero también lo podríamos haber hecho por hora. Guardaremos un mapa de (fecha -> cuenta) en un mapa en memoria cuya llave será el “hashtag”. Como todo ésto se ejecuta en paralelo y de forma incremental, tenemos que poder mezclar dos mapas potencialmente diferentes en un momento dado (una operación de “combine” como las de Hadoop). Fijaros especialmente en la operación combine():

public static class CountByDate implements CombinerAggregator<Map<String, Long>> {

  public Map<String, Long> init(TridentTuple tuple) {
    Map<String, Long> map = zero();
    map.put(tuple.getString(1), 1L);
    return map;
  }

  public Map<String, Long> combine(Map<String, Long> val1, Map<String, Long> val2) {
    for(Map.Entry<String, Long> entry : val2.entrySet()) {
      val2.put(entry.getKey(), MapUtils.getLong(val1, entry.getKey(), 0L) + entry.getValue());
    }
    for(Map.Entry<String, Long> entry : val1.entrySet()) {
      if(val2.containsKey(entry.getKey())) {
        continue;
      }
      val2.put(entry.getKey(), entry.getValue());
    }
    return val2;
  }

  public Map<String, Long> zero() {
    return new HashMap<String, Long>();
  }
}

Ahora veremos cómo la capa de servicio y la de tiempo real se interconectan en el servicio DRPC.

El servicio DRPC

Una de las cosas buenas de Storm es que puede ejecutar llamadas RPC de forma distribuída. Podemos usar ésto para crear un servicio que estará detrás de un website y que mostrará evoluciones temporales para “hashtags”. Ahora que tenemos una capa “batch” que computa todo el histórico de las estadísticas y las vuelca a Splout SQL, y una capa de “tiempo real” que puede calcular la evolución en tiempo real de un “hashtag”, ¿Cómo juntamos todo ésto?

Vamos a añadir un servicio DRPC a la topología Storm:

topology
  .newDRPCStream("hashtags", drpc)
  .each(new Fields("args"), new Split(), new Fields("hashtag"))
  .groupBy(new Fields("hashtag"))
  .stateQuery(hashTagCounts, new Fields("hashtag"), new MapGet(), new Fields("resultrt"))
  .stateQuery(sploutState, new Fields("hashtag", "resultrt"), new HashTagsSploutQuery(), new Fields("resultbatch"))
  .each(new Fields("hashtag", "resultrt", "resultbatch"), new LambdaMerge(), new Fields("result"))
  .project(new Fields("result"));

Las queries RPC se paralelizan por “hashtag”. Para cada una de ellas, dos queries se ejecutan en secuencia: una a la capa “tiempo real” (hashTagCounts) y la otra a la capa de “servicio” con los datos de la capa “batch”: (sploutState). Fijaros cómo usamos la función Trident MapGet() para poder obtener los resultados de la capa “tiempo real” y la función que hemos programado antes HashTagsSploutQuery() para obtener los datos de Splout SQL. Fijaros que la tupla al final tiene tres campos: el hashtag, el resultado “batch” y el resultado “tiempo real”. Al final, usamos una función llamada LambdaMerge() que mezcla el resultado de ambas capas para devolver una respuesta unificada al usuario del servicio DRPC.

La lógica de negocio para mezclar los resultados de ambas capas es muy sencilla. Simplemente asumimos que si un resultado viene de la capa “batch” ese resultado tiene prioridad (por lo tanto, “machaca” cualquier posible resultado de la capa “tiempo real”). Además, devolvemos un mapa ordenado por la llave:

public static class LambdaMerge extends BaseFunction {

  public void execute(TridentTuple tuple, TridentCollector collector) {
    Map<String, Long> resultRealTime = (Map<String, Long>) tuple.get(1);
    QueryStatus resultBatch = (QueryStatus) tuple.get(2);
    TreeMap<String, Long> consolidatedResult;

    if(resultRealTime != null) {
      consolidatedResult = new TreeMap<String, Long>(resultRealTime);
    } else {
      consolidatedResult = new TreeMap<String, Long>();
    }
    if(resultBatch != null) {
      if(resultBatch.getResult() != null) {
        for(Object rowBatch : resultBatch.getResult()) {
          Map<String, Object> mapRow = (Map<String, Object>) rowBatch;
          String day = (String) mapRow.get("day");
          Long count = Long.parseLong(mapRow.get("SUM(count)").toString());
          consolidatedResult.put(day, count);
        }
      }
    }
    collector.emit(new Values(consolidatedResult));
  }
}

Probarlo

Podeis seguir las instrucciones en github (en inglés) para probar este sencillo ejemplo. El ejemplo usa una sucesión infinita de dos falsos tweets en tiempo real (“#california is cool”, “I like #california”) y hace queries a Splout SQL donde se debería haber cargado el conjunto de datos de ejemplo – que también contienen el hashtag “california”. Así que después de ejecutarlo todo se debería ver algo como:

...
Result for hashtag 'california' -> [[{"20091022":115,"20091023":115,"20091024":158,"20091025":19}]]
Result for hashtag 'california' -> [[{"20091022":115,"20091023":115,"20091024":158,"20091025":19,"20130123":76}]]
Result for hashtag 'california' -> [[{"20091022":115,"20091023":115,"20091024":158,"20091025":19,"20130123":136}]]
Result for hashtag 'california' -> [[{"20091022":115,"20091023":115,"20091024":158,"20091025":19,"20130123":192}]]
Result for hashtag 'california' -> [[{"20091022":115,"20091023":115,"20091024":158,"20091025":19,"20130123":232}]]
Result for hashtag 'california' -> [[{"20091022":115,"20091023":115,"20091024":158,"20091025":19,"20130123":286}]]
...

Como se vé, la última fecha (que debería coincidir con la fecha de hoy) se incrementa en tiempo real mientras las otras fechas históricas no evolucionan – se han añadido a la respuesta a través de los resultados de la capa “batch” provistos por Splout SQL.

Conclusiones

A través de este ejemplo hemos visto Trident, una API interesant de alto nivel por encima de Storm, y Splout SQL, una base de datos SQL raṕida, particionada y de sólo lectura para Hadoop. Hemos visto un ejemplo real de una “arquitectura lambda”, omitiendo ciertas partes que no eran tan interesantes para éste ejemplo.

Pero no podemos acabar sin mencionar algo que nos hemos dejado: hablar de un “coordinador”. Debería haber un coordinador que lanzara el proceso “batch” periódicamente y además asegurar que la capa “batch” sólo procesa datos completos. Con datos completos nos referimos a que si la agrupación mínima es por hora, la capa “batch” no debería procesar horas incompletas (éstas quedarían sólo para la capa “tiempo real”). De esta manera, se cumpliría la propiedad de que cualquier estadística calculada por la capa “batch” puede “machacar” la estadística correspondiente en la capa “tiempo real”. Por otro lado, para que la capa “tiempo real” sea eficiente, debería expirar datos antiguos periódicamente. Pero hay que tener cuidado, ya que si se expiran los datos antes de que la capa “batch” haya calculado lo correspondiente, se corre el riesgo de tener “agujeros” en los resultados.

Como hemos dicho varios veces, aquí teneis el código completo con comentarios en inglés.

Entrada original en: Un ejemplo de “arquitectura lambda” para el análisis en tiempo real de hashtags usando Trident, Hadoop y Splout SQL

Motivación

La motivación para haber creado Splout SQL está bien explicada en este post.

Los principios de Splout SQL

Splout SQL es una base de datos escalable, tolerante a fallos, de sólo lectura, particionado, RESTful y SQL que se integra muy bien con Hadoop. Splout SQL se puede usar para ejecutar queries sobre datasets generados por Hadoop bajo alta carga, en tiempos de respuesta muy bajos. Ésto significa que Splout SQL es adecuado para escenarios como aplicaciones web o móviles que requieren un alto rendimiento. Por ejemplo, hemos usado Splout para servir 350GB de datos de los “Wikipedia Pagecounts“, ejecutando agrupaciones arbitrarias en tiempo real sobre ellos (ver el benchmark para más detalle).

La integración con Hadoop sigue los mismos principios de diseño que la de Elephant DB (que están muy bien explicados en éstas transparencias). A modo de resumen, los datos son indexados y particionados “offline” en un clúster Hadoop. Las estructuras de datos generadas son entonces movidas a un clúster de servicio de sólo lectura, que es el que atiende las queries de los usuarios.

Elephant DB hace algo parecido. Voldemort también. Inspirados por ellos, hemos recopilado conceptos como “hot-swapping”, “rollback”, versiones (“domains” / “stores” / “tablespaces”) en una API sencilla y fácil de usar para desplegar datos de servicio en modo de sólo lectura.

Pero Splout puede crear tablas arbitrarias con B-Trees, ofreciendo una vista SQL real sobre el dataset final. Por lo tanto Splout SQL es una extensión más rica y potente que las bases de datos “llave / valor” que hemos mencionado antes, ya que permite agregaciones SQL en tiempo real arbitrarias sobre las vistas generadas.

Cómo funciona

Las estructuras de datos necesarias para contestar queries son generadas usando los métodos y programas de “splout-hadoop”. Ésta API usa Pangool para ejecutar un proceso batch que calcula la distribución de los datos y los indexa. Los ficheros de datos generados no son más que ficheros binarios SQLite y se guardan en el HDFS o cualquier otro sistema de ficheros compatible con Hadoop (por ejemplo S3). Éstos ficheros entonces son descargados desde un clúster diferente, el de servicio, que ejecuta los servicios de Splout SQL, bajo una petición de “deploy”. Éste deploy distribuído y asíncrono es coordinado por Hazelcast hasta que triunfa o falla.

En el lado del servicio, las queries de los usuarios son enrutadas a través de un servicio llamado “QNode”, que conoce la distribución de los datasets que sirve. Los servicios “DNode” son los que tienen los ficheros SQL, contestan las queries y mandan el resultado hacia arriba. El resultado de la query es finalmente devuelto al usuario como un JSON a través de una interfaz HTTP.

Éstas son algunas de las peculiaridades del sistema:

• El sistema es capaz de distribuír los datos de forma equitativa a sus “shards” usando técnicas de sampleo (como la que se usa en el TeraSort o bien algoritmos más sofisticados como Reservoir Sampling).
• Para poder samplear y distribuír los datos debe haber algún tipo de “llave de particionamiento“. En Splout, los datasets pueden ser particionados por cualquier función aplicable a una fila de datos. Normalmente ésto será una o varias columnas del conjunto de datos. Esta función permite al proceso de sampleo determinar el histograma de los datos y distribuírlos equitativamente sobre el número de particiones deseado.
• Las queries siempre van a una única partición, así que el usuario debe proveer la “llave de particionamiento” para cada query. Esta llave debe ser, por supuesto, consistente con la función de particionado provista anteriormente. (Esto es lo mismo que dar una “llave” para obtener un “valor” en una base de datos “llave / valor”, sólo que en este caso obtenemos el resultado de una query SQL como valor).
• Se obtiene “fail-over” en el sistema a través de un mecanismo sencillo de replicación: cada partición se copia a múltiples DNodes para que así, si uno de ellos cae, los otros pueden todavía servir esa partición.
• El usuario puede crear cualquier número de tablas dentro de “tablespaces”, y desplegar un número arbitrario de tablespaces a la vez. Las tablas pueden ser particionadas o replicadas a todas las particiones (por ejemplo, tablas pequeñas que pueden ser replicadas en cada partición para así poder ejecutar joins con ellas).
• Los “deploys” son atómicos y no afectan el servicio de las queries, así que el sistema como conjunto es bastante robusto, y a la vez sencillo y flexible. Además, el usuario puede volver a versiones anteriores de los datos desplegados de forma atómica también (“rollback”).

Splout SQL en el espacio “Big Data”

Hemos diseñado Splout SQL para cubrir un hueco en el espacio “Big Data”. Pensamos que Splout es el mejor candidato a ser la “Batch Layer” de la “Lambda architecture“.

Splout SQL se integra bien con Hadoop, puede contestar queries a muy baja latencia y está preparado para servir datos que exceden en mucho la memoria RAM disponible, como Voldemort o Elephant DB, pero ofrece agregaciones SQL en tiempo real en vez de ser una simple base de datos “llave / valor”.

Por otro lado, las bases de datos NoSQL como MongoDB, Cassandra y HBase, a pesar de no ser SQL, ofrecen un gran abanico de “features”, pueden ser usadas junto con Hadoop y parecería que comparten muchos de los objetivos de Splout SQL. Sin embargo, son sistemas mucho más complicados de manejar y usar. Y que sepamos, en el momento en que escribimos este documento, ninguno de ellos provée un puente con Hadoop que sea 1) fácil de usar y manejar 2) que no afecte el servicio de los datos 3) que sea atómico y ofrezca la posibilidad de “rollback”.

Finalmente, hay un conjunto de sistemas de “analítica rápida” que están surgiendo por encima de Hadoop que son capaces de contestar queries SQL (Impala, Apache Drill). Pero, que sepamos, han sido concebidos para ser usados “offline”, en aplicaciones de analítica. No han sido pensados para soportar entornos con alta carga, mucho tráfico, ni para devolver queries en muy bajas latencias (por debajo del segundo siempre). Lo que hacen es que aprovechan técnicas como el almacenamiento columnar para poder escanear muchos datos muy rápidamente (en el orden de los segundos), pero no tienen estructuras de datos como B-Trees para poder acceder a cualquier parte del dataset muy rápidamente (en el orden de los milisegundos). Así que, en general, son adecuados para unas cuantas queries que analizan todo el conjunto de datos, mientras que Splout es más bien adecuado para muchas queries que impactan porciones muy pequeñas del conjunto de datos.

Pruébalo!

Splout todavía no es 1.0 pero nos gustaría que lo probaras y nos dieras feedback, así que ¡dale!

Entrada original en: Presentando Splout SQL: Un conector SQL de muy baja latencia para Hadoop

Es un placer para nosotros el compartir las siguiente presentación usada en las conferencias:

Entrada original en: Tuple MapReduce en el IEEE ICDM 2014

La siguiente presentación contiene una introducción sobre el Big Data y más específicamente sobre MapReduce. Posteriormente, se describen casos de usos donde MapReduce ha demostrado su utilidad y finalmente se presenta el nuevo paradigma Tuple MapReduce y su implementación en Hadoop: Pangool

Los casos de uso reales de MapReduce son los siguientes:

• Data analytics
• Crawling
• Full-text indexing
• Reputation systems
• Data mining

Entrada original en: Casos de uso reales de MapReduce y Hadoop

Aquí tenéis las transparencias disponibles para que lo veáis con más detalle.

Entrada original en: Extracción de valor de transacciones con tarjeta de crédito