Spark dispone de un módulo para el procesamiento de consultas SQL. Estas consultas pueden utilizarse para manipular datos estructurados de lo más variado (JSON, CSV, Parquet, etc.) y sobre multitud de repositorios (bases de datos relacionales y no relacionales, Amazon S3, Hadoop, etc.). Además, es compatible con HiveQL, un lenguaje similar al conocido SQL que se ha convertido en indispensable a la hora de realizar operaciones sobre grandes cantidades de datos en plataformas Hadoop.

En este ejemplo, veremos cómo realizar consultas SQL sobre documentos en formato JSON. Para ello puedes descargar el fichero personas.json, donde en cada línea encontrarás la representación de una entidad Persona con los atributos nombre, edad y género. Es importante recalcar que en cada línea se encuentra una única representación de la entidad Persona en JSON, ya que de lo contrario el fichero no podría ser procesado por Spark. Debes copiar este fichero en "C:", para poder hacer referencia al mismo en el código del programa (línea 19) que leerá las personas contenidas en dicho fichero.

Para tener acceso al fichero y poder realizar las consultas, se crea un HiveContext. Esta clase dispone de métodos para leer archivos de diferentes tipos, en este caso JSON. Se puede especificar un fichero o un directorio que contiene ficheros JSON. En  nuestro caso será un único fichero. La lectura de este fichero devuelve una estructura de tipo DataFrame. Este tipo de dato es similar al utilizado en R y viene a representar los datos como si de una tabla de base de datos se tratara. El esquema de los datos y su tipología se infieren automáticamente por Spark, aunque para ganar en eficiencia pueden ser especificados de forma manual.

Una vez que tenemos el DataFrame, se creará una tabla temporal cacheada (líneas 22 y 23) sobre la que poder lanzar las consultas SQL. Una primera consulta obtiene todas las personas del fichero, y una segunda obtiene el número de personas del fichero. Los datos se devuelven en una estructura de tipo DataFrame.

A continuación hemos utilizado un método de agregación proporcionado por el DataFrame. Este método devuelve el máximo, mínimo y número de personas en un DataFrame.

Podemos visualizar los resultados con la función show() que dibujará por consola los DataFrame.

Pulsa aquí para acceder al código de esta entrada en mi repositorio GitHub

En esta segunda entrada sobre Apache Spark veremos como ejemplo el "Hello world" del mundo Big Data: contar las palabras de un texto y ver qué palabras son las más utilizadas en el mismo. Para ello, utilizaremos un archivo de texto que contiene todas las obras de William Shakespeare descargable desde aquí (aunque ya lo incluyo en el proyecto que puedes clonar desde mi repositorio GitHub).

Es importante señalar que para poder probar este ejemplo, necesitarás descargar un archivo llamado winutils.exe en este enlace, y copiarlo en la ruta "C:\hadoop\bin", de lo contrario la ejecución del código dará el mensaje de error Failed to locate the winutils binary in the hadoop binary path. Además, debes copiar el archivo shakespeare.txt en "C:".

En el código, hemos definido en la línea 20 una función para eliminar espacios en blanco adicionales, signos de exclamación, interrogación, puntuación, etc. con el fin de limpiar el fichero de texto. Esta función la hemos aplicado sobre el texto en la línea 30 utilizando la función map. 

Posteriormente obtenemos por cada línea las palabras y eliminamos aquellas que estén en blanco (línea 36) utilizando las funciones flatMap y filter. En la línea 38 obtenemos el número de palabras totales.

Ahora vamos a contar las apariciones de cada palabra en el texto, para lo cual utilizaremos el paradigma mapreduce mediante las funciones que nos ofrece Apache Spark. Para ello, convertiremos cada palabra en el par (palabra, 1) mediante la función map (línea 41). Posteriormente, utilizaremos la función reduceByKey para agrupar estos pares en base a su clave, que es la palabra, y contabilizar el número de apariciones de cada una sumando el valor de cada par (línea 42).

Para finalizar, imprimiremos por pantalla las 50 palabras que más se repiten en el texto. Para ordenar las palabras de mayor a menor número de apariciones, hemos utilizado en la línea 50 la función shortWith, que ordena en base al criterio establecido los pares (palabra, num. apariciones).

Pulsa aquí para acceder al código de esta entrada en mi repositorio GitHub

Estamos en la era de los datos o el denominado Big Data. Disponemos de tal cantidad de datos, que los tradicionales sistemas de almacenamiento, gestión y procesamiento de los mismos ya no son válidos. Es por ello que en los últimos años han aparecido multitud de herramientas y frameworks para gestionar grandes cantidades de datos de forma eficiente. Una de estas herramientas es Apache Spark.

En esta entrada hablaré de este framework de procesamiento y análisis distribuido de datos, mostrando una introducción básica al mismo. En siguientes entradas mostraré más funcionalidades con ejemplos más complejos.

Para seguir esta entrada se necesitan las siguientes herramientas.

• JDK 7
• IDE Scala
• Spark 1.6 para Hadoop 2.6

¿Qué es Apache Spark?

Spark es un framework de procesamiento y análisis distribuido de grandes cantidades de datos en memoria. Dispone de diversos módulos que permiten aplicar operaciones sobre datos, realizar consultas, procesamiento de grafos, stream processing y aplicación de algoritmos de machine learning de una forma eficiente y escalable. Al ser un entorno de computación en memoria, Spark puede conectarse a repositorios de datos externos como HBase, Cassandra, MongoDB, etc. donde residen datos persistentes.

Spark está escrito en Scala, un lenguaje de programación funcional soportado por la máquina virtual Java (JVM). De esta forma, el código Scala se compila y se convierte en bytecode Java, por lo que un programa escrito en Java es perfectamente interoperable con otro escrito en Scala.

El framework Spark dispone de una API para poder implementar aplicaciones de procesamiento y análisis de datos. Esta API se encuentra soportada por lenguajes de programación como Java o Python, aunque lo más natural es utilizar Scala, ya que es el lenguaje de programación que sustenta Spark.

Arquitectura de Spark

Los principales componentes de Spark son los que se describen a continuación:

• Workers: son los nodos donde se ejecutan de forma distribuida las operaciones o tareas de las aplicaciones Spark.
• Cluster manager: gestiona los workers. Spark dispone de su propio manager, aunque también soporta los gestores Mesos y YARN.
• Driver program: es la aplicación principal. Las diferentes tareas que se implementen en esta aplicación serán las que se distribuyan por los workers del clúster para la realización en paralelo de las diferentes operaciones requeridas.

API de Spark

Antes de entrar en materia y poner un sencillo ejemplo de procesamiento de datos con Spark, es conveniente entender algunos conceptos básicos de la API.

• SparkContext: facilita la conexión de la aplicación con un clúster Spark.
• Resilient Distributed Dataset (RDD): un RDD es una colección de elementos que puede particionarse para ser tratados en paralelo. La API proporciona métodos para conectarse a diferentes repositorios de datos (Hadoop, Cassandra, etc.) y crear RDDs. La creación de un RDD es una operación "lazy", es decir, que Spark no realiza ninguna acción al momento, sino que realiza una programación de dicha operación hasta que ocurra alguna acción sobre los datos del RDD. Las operaciones que pueden realizarse sobre un RDD son transformaciones y acciones.
• Transformaciones: una operación de transformación sobre un RDD devuelve un nuevo RDD al que se le aplica cierta operación. Por ejemplo, aplicar una determinada función a los elementos del RDD (p. ej. multiplicar por 2 cada elemento), filtrar los elementos en base a una condición determinada, etc.
• Acciones: las acciones devuelven un valor a la aplicación o Driver program. Por ejemplo, contar los elementos de un RDD, calcular el máximo, el mínimo o guardar el contenido de un RDD en un fichero son ejemplos de acciones sobre un RDD.

Ejemplo con Apache Spark

Para empezar a trabajar con Spark necesitaremos descargar y descomprimir el IDE Scala, que está basado en Eclipse. Para lanzar el entorno ejecutaremos el archivo eclipse.exe. Después, descargaremos el proyecto con el código fuente del repositorio GitHub en este enlace.

Ahora veremos un ejemplo muy sencillo en el que filtraremos una lista de números para obtener los que sean pares. En primer lugar crearemos un objeto llamado NumerosPares que extienda de App (línea 6) dentro del fichero de nombre NumerosPares.scala. Con esto, estamos definiendo una nueva aplicación de Scala, que ejecutará el código contenido en el propio objeto.

Configuraremos una conexión a un clúster Spark, en este caso local, mediante el objeto SparkConf, que después utilizaremos para instanciar un objeto de tipo SparkContext. Después crearemos un RDD partiendo de una lista de 10.000 números. Sobre este RDD de nombre listaNumerosRdd, realizaremos una transformación mediante un filtro. El filtro aplicará una función para obtener un nuevo RDD que solamente contenga los números pares. La notación utilizada en la función mediante la barra baja "_" es equivalente a escribir { x => x % 2 == 0} (línea 15).

Finalmente aplicaremos una acción sobre el RDD, para contar el número de elementos pares que después visualizaremos por consola.

Para ejecutar la aplicación, nos posicionaremos con el ratón en el fichero NumerosPares.scala, pulsando con el botón derecho del ratón y seleccionando Run As -> Scala Application.

Conclusiones

Spark es un entorno con múltiples funcionalidades, que simplifica enormemente la gestión de grandes cantidades de datos de forma transparente. La magia de Spark reside en que el programa escrito en el ejemplo, puede ejecutarse tanto en un clúster local, como en un clúster distribuido compuesto por multitud de nodos sin cambiar ni una sola línea de código.

En siguientes entradas veremos ejemplos más completos de la API Spark para la gestión de grandes cantidades de datos, así como otros módulos ofrecidos por el framework como los de consulta de datos o el de aplicación de algoritmos de machine learning.

Pulsa aquí para acceder al código de esta entrada en mi repositorio GitHub