Spark (I) - Introducción

Spark (II) - Características principales

Spark (III) - ¿Cómo crear y configurar un Clúster?

Spark (IV) - Desarrollando programas en Spark. Conceptos Previos.

Spark (V) - Desarrollando programas en Spark. Empezando a programar.

Spark (VI) - Monitorización de Spark

Spark (VII) - Ecosistema alrededor de Spark

Antes de comenzar a meter las manos en Unix tenemos que saber de que está compuesto un clúster Spark.

1. Las aplicaciones de Spark se ejecutan en sets de procesos independientes en el cluster y que a su vez son coordinados por el objecto SparkContext del programa principal (llamado driver program). 

2. El SparkContext puede conectarse a muchos tipos de clúster managers (tanto Spark StandAlone como los administrador por Mesos/YARN), el cual asigna recursos a través de las aplicaciones. Una vez conectado, Spark toma los executors en los nodos del cluster, los cuales son "worker processess" que ejecutan operaciones y almacenan los datos de la aplicación.

3. Después se envía el código de tu aplicación (definido por el archivo JAR o Python  que hemos enviado al SparkContext) a los executors. 

4. Por último, el SparkContext envía las tareas a los executors para ejecutarlas.

Hay algunas cosas útiles que debemos tener en cuenta:

1. Cada aplicación obtiene su propio proceso executor, el cual está levantado durante toda la ejecución y que además ejecutará tareas en múltiples threads. De esta manera las aplicaciones se encuentran aisladas unas de las otras, tanto del lado del scheduler (cada driver maneja sus propias tareas) como del lado del executor (las tareas de las diferentes aplicaciones se ejecutan en diferentes JVMs). Sin embargo, esto también significa que los datos no pueden ser compartidos entre diferentes aplicaciones de Spark (instancias del SparkContext) sin que estas tengan que escribir en sistemas de almacenamiento externo.

2. Spark es agnostico al tipo de administrador de cluster que hayamos elegido. Puede soportar tanto MESOS como YARN.

3. Dado que el driver administra y prioriza las tareas del cluster, este debería ser ejecutado cerca de los workers (nodos), preferiblemente en la misma red local (por temas de latencia).

1. De manera sencilla, en donde hemos extraído Spark ejecutamos:

bin/spark-shell

Esto nos levantará un shell interactivo en Scala con un SparkContext para jugar con Spark.

En este punto es necesario recordar que si queremos utilizar HDFS necesitaremos utilizar una versión de Hadoop acorde a la versión de Spark que estemos utilizando. En Spark 0.9.1 se utiliza de Apache Hadoop es 1.2.1.

Nota: no ansiarse malandrines, en próximos capítulos jugaremos con el Shell y haremos programillas. Mientras me esperáis, podéis echar un ojo a la docu oficial que esta genial:  http://spark.apache.org/docs/latest/

Modo Standalone

1. Iniciamos el Master (desde el directorio donde extraímos Spark):

sbin/start-master.sh

Una vez se ha iniciado, abrimos un navegador y vamos a http://localhost:8080. Veremos una interfaz web un poco parca:


Antes de continuar tenemos que saber que para que haya comunicación entre el maestro de Spark y los esclavos tenemos que tener como dicen por ahí "passwordless" entre ellos. Si queremos conectarnos entre localhost y localhost tendremos que superar esta barrera:

2. Generamos la clave privada, para ello ejecutamos en el terminal:

ssh-keygen

Nota: presionad enter hasta que deje de preguntar. Por defecto se guardará en el directorio ~/.ssh con nombre id_rsa.pub

3. Esta clave la copiamos como parte pública a ~/.ssh/authorized_keys. Primero instalamos OpenSsh-server si no lo tenemos instalado:

sudo apt-get install openssh-server

ssh-copy-id [TU_NOMBRE_USUARIO]@[HOSTNAME_DE_TU_MÁQUINA]

Ejemplo:

ssh parallels@parallels-Parallels-Virtual-Platform

Escribimos yes y si vamos a la carpeta ~/.ssh, veremos que se ha generado un fichero llamado authorized_keys.

3. Si queremos añadir workers al cluster standalone, añadimos el número con el que queremos jugar al spark-env.sh:

echo "export SPARK_WORKER_INSTANCES=3" >> ./conf/spark-env.sh

4. Por último, para levantar los workers ejectuamos:

sbin/start-slaves.sh

Ahora volvemos a la interfaz web http://localhost:8080 y veremos que además del master ya se encuentran levantado los workers:


Llegados a este punto conviene saber cómo paramos el cluster, cómo lo iniciamos, etc:

· Parar clúster:

sbin/stop-all.sh

· Iniciar clúster:

sbin/start-all.sh

· Parar máster:

sbin/stop-master.sh

· Iniciar máster:

sbin/start-master.sh

· Parar esclavos:

sbin/stop-slaves.sh

· Iniciar esclavos:

sbin/start-slaves.sh

Modo Spark + Amazon Ec2

Otro modo que tenemos para lanzar Spark y que automágicamente tengamos un Spark con Workers distribuidos es mediante su script de Ec2.

El script se encuentra en el directorio ec2 y nos permitirá lanzar, administrar y para clusters de Spark para Amazon Ec2.

Nos levanta de manera automática Spark, Shark y HDFS. Cada cluster se identifica por las máquinas que haya en los grupos de seguridad de EC2 y esos nombres serán los que tengan los componentes del cluster. Por ejemplo, un cluster llamado test que contiene un nodo master contendrá un grupo de seguridad llamado test-master, y los nodos esclavos estarán en test-slaves.

Requisitos previos de Amazon EC2
Antes de lanzar el Script tenemos que tener en cuenta algunas cosillas de Amazon:

· Necesitamos crear un par de claves Amazon EC2. Esto se puede hacer desde la consola AWS en la sección de Key Pairs. Una vez creada la guardamos en nuestro local con permisos 600.

· Para poder utilizar el script de spark-ec2 necesitaremos tener las siguientes variables de entorno bien configuradas:

export AWS_ACCESS_KEY_ID=[la clave de identificación de AWS] Normalmente alojada en Credenciales de seguridad de AWS.

export AWS_SECRET_ACCESS_KEY=[el password]

Lanzando el Cluster

· Donde:
<keypair> : el nombre del key pair que establecimos anteriormente con la extensión incluida.
<key-file> : el lugar donde está el archivo key pair que descargamos anteriormente.
<num-esclavos> : el número de esclavos que queramos que tenga el cluster. Debemos darnos cuenta que son sólo los esclavos. Esto quiere decir que se nos creará una instancia para el máster  + una intancia para cada nodo esclavo que establezcamos aquí.
<nombre-cluster> : el nombre del cluster, que indicamos que era importante para el nombre de los grupos de seguridad.

Para más opciones de lanzamiento del cluster visita este link.

Ejecutar Aplicaciones
· Vamos al directorio ec2.

· Para copiar nuestra aplicación podemos utilizar el script copy-dir (dentro de la carpeta ec2), que nos hará un RSYNC al mismo sitio en todos los esclavos.

· Si la aplicación necesita un gran dataset, la mejor manera de hacer esto es mediante los buckets de S3 o instanciando un EBS con los datos ya cargados en los nodos que haya instancias de Hadoop. Al ejecutar el script spark-ec2 este nos instala un HDFS en todas las máquinas. Aquí cuidado, porque lo hace en instancias efímeras (ephemeral-hdfs).

Aquí está bien parar y ver la diferencia entre persistent-hdfs y ephemeral-hdfs:

- Persistent-hdfs: todo aquello que guardemos aquí se mantendrá aunque reiniciemos o apaguemos las máquinas. Por defecto el espacio persistente es pequeño (unos 3 Gb pone en la docu oficial) aunque siempre podemos aumentarlo al crear el cluster mediante el comando --ebs-vol-size y poniendo la cantidad de gigas que consideremos.

- Ephemeral-hdfs: todo aquello que guardemos aquí se perderá tanto al reiniciar como al apagar las máquinas.

Para más información sobre Amazon Ec2 visita este link.

Modo Spark + Mesos

Aquí he estado pensando en hacer el tutorial pero he visto que en la web de Mesos lo tienen genial y creo que mucho mejor no lo iba a hacer. Por tanto os recomiento seguir este tuto:

Hoy veremos las características principales que hacen a Spark tan atractivo.

Resilient Distributed Dataset

En Spark a diferencia de Hadoop, no utilizaremos una colección de datos distribuidos sino que usaremos los RDDs (Resisilient Distribuited Datasets). Normalmente no suelo dar la chapa sobre en que consiste esto a bajo nivel, pero ya que estamos en el blog lo dejo aquí por escrito:

Son colecciones lógicas, inmutables y particionadas de registros a lo largo del cluster que pueden ser reconstruidas si alguna partición se pierde (no necesitan ser materializadas pero si reconstruidas para mantener el almacenamiento estable). Se crean mediante la transformación de datos utilizando para ello transformaciones (filtros, joins, Group by...). Por otra parte permite cachear los datos mediante transformaciones como Reduce, Collect, count, save...

Los RDD son tolerantes a fallos, para ello mantiene una cosa llamada el "lineage" del RDD. Este "lineage" permite que los RDDs se reconstruyan en caso de que una porción de datos se pierda.

Por tanto, los beneficios de utilizar RDD´s son varios:

· La consistencia se vuelve más sencilla gracias a la inmutabilidad.

· Tolerancia a fallos con bajo coste (gracias al "lineage" mantenemos checkpoints de los datos).

· A pesar de ser un modelo restringido a una serie de casos de uso por defecto, gracias a RDDs se puede utilizar Spark para un cantidad de aplicaciones muy variadas.

Suelo contar que yo me imagino los RDD como sacos de datos en memoria en los que si se pierde parte de los datos del saco, el saco se adapta a la cantidad de datos que tenga dentro.

Operaciones en RDDs

Análisis Interactivo de Datos

Una de las características más útiles es el Shell interactivo que permite a un usuario utilizar las capacidades de Spark inmediatamente (no se necesita un IDE o compilar el código). La Shell es una herramienta que se utiliza para la exploración de datos interactivamente, que nos permitirá probar porciones de la aplicación que estemos desarrollando.

Spark puede leer y escribir datos desde y hasta HDFS. Lo bueno es que nos permite jugar con APIs en Java, Scala y Python para aplicaciones "stand-alone".

Batch Processing más Rápido

Los primeros despliegues de Spark se enfocaron en aumentar el rendimiento de las aplicaciones MapReduce existentes. Recodemos que MapReduce en la actualidad es un framework genérico de ejecución paralela de aplicaciones y no es exclusivo ni una implementación propietaria del core de Hadoop. Por lo tanto Spark proporciona también MapReduce que puede llegar a ser más eficiente gracias al uso de la memoria (mientras utilizará "lineage" para recuperar los fallos en caso de que fuese necesario).

Algoritmos Iterativos

Spark permite a los usuarios y a las aplicaciones especificar el cacheo de un dataset mediante la operación cache(). Esto significa que nuestra aplicación puede acceder a la memoria RAM en vez de al disco, lo cual aumenta de manera espectacular el rendimiento en algoritmos iterativos ya que pueden acceder acceder al mismo dataset repetidamente. Ese caso de uso cubre un gran número de aplicaciones, como pueden ser las aplicaciones orientadas a machine learning o los algoritmos gráficos que son iterativos por naturaleza.

Puedes ver más sobre esto en la web oficial de Spark.

Procesamiento de Streams en Tiempo Real

Dado que las operaciones realizadas los sistemas de análisis de datos poseen una latencia baja, es natural extender el motor hacia streams de datos en tiempo real. Spark tiene un API que nos permite trabajar con estos streams de datos y que además tienen una ventaja diferencial: podemos mezclar streams en tiempo real con programas Batch.

http://www.berkeley.edu/index.html
http://spark.apache.org/