Procesamiento de Datos en Streaming usando Kafka y Spark Structured Streaming
Vivimos actualmente en un mundo donde se genera una gran cantidad de datos por segundo, las industrias y los casos de uso son variados. Esta gran cantidad de datos en streaming pueden proveer información útil si es que se analiza correctamente, o con la ayuda de algoritmos de aprendizaje automático predecir patrones de comportamiento. E.g. Entre las empresas que usan procesamiento de datos en Spark están Amazon, Netflix, Uber. Apache Spark ha evolucionado rápidamente siendo la tecnología Big Data mayormente usada cuando nos referimos a procesamiento de datos en streaming. La ingesta de datos en streaming se puede hacer desde muchas fuentes como Apache Flume, Kinesis, Kafka, NATS o sockets TCP. Particularmente usando Spark Structured Streaming, es posible el procesamiento de algoritmos complejos expresados como funciones de alto nivel como map, reduce, join y window. El procesamiento de datos en streaming usando Spark es una de las herramientas más potentes que sirve para solucionar casos de uso complejos y puede ser fácilmente ser integrados con las APIs de Spark Core como SparkSQL, SparkML, y GraphX.
Haciendo un breve paréntesis, inicialmente se hablaba de que Spark Streaming era realmente un microbatch, pero ahora con el uso de Spark Structured Streaming en la versión 2.x de Spark ya se habla de procesamiento de datos en tiempo real con el uso de ventanas (windows) de tiempo en los cuales se procesa la data.
Si bien es cierto el procesamiento de datos en Streaming ahora se puede hacer en la nube de manera serverless en la nube o como un servicio PaaS, el día de hoy me centrare en una simulación de un entorno on-premise. Con lo cual necesitaremos implementar Kafka, Spark. Estos elementos también se pueden instalar en contenedores ya sea on-premise o en la nube.
Hacer procesamiento datos usando Kafka y Spark Structured Streaming es un clásico en streaming IoT o procesamiento en datos que vienen de Redes Sociales.
Vamos a simular la generación de datos en Streaming. Para ello tomaremos datos de un CSV y cada linea será enviada al “Producer” de Kafka y luego desde la API del “Consumer” de Kafka vamos a tomar los datos y los procesaremos usando Spark Streaming para finalmente mostrarla en consola.
La arquitectura utilizada es la siguiente:
Pasos a seguir:
- Iniciar una máquina virtual en AWS con RedHat, t2.micro Free-tier. Este será utilizada para el “Productor de Datos”
- Iniciar una segunda máquina virtual en AWS similar a la anterior con Kafka, ver articulo. Este sera utilizado para nuestro “Broker de Datos”
- Iniciar una tercera maquina virtual en AWS similar a las anteriores con Spark, ver articulo. Este será utilizado para el “Consumer de Datos”.
- Asegurarse que se tenga conexión de red entre estas máquinas virtuales
- Iniciaremos la producción de datos, pasando por el bus Kafka y luego agregando los datos en streaming usando Structured Streaming de Spark
Paso 1 — Productor de Datos en Streaming
En esta maquina virtual traeremos los CSVs que contienen la data, esta data contiene cuatro columnas “ID del sensor”, “Velocidad”, “Aceleración en el eje X”, “Aceleración en el Y”, “Aceleración en el Z”. Vamos a simular data de sensores IoT.
Instalaremos wget para traer los archivos CSV a nuestra máquina virtual:
$ sudo yum install wget# Descargamos desde Github los archivos CSVs
--------------------------------------------$ wget https://raw.githubusercontent.com/mtpradoc/kafka-spark-demo-pyconsg19/blob/master/iot_demo/data/iot_data_id10.csv$ wget https://raw.githubusercontent.com/mtpradoc/kafka-spark-demo-pyconsg19/blob/master/iot_demo/data/iot_data_id20.csv$ wget https://raw.githubusercontent.com/mtpradoc/kafka-spark-demo-pyconsg19/blob/master/iot_demo/data/iot_data_id30.csv# Descargamos desde Github el código en Python que vamos a usar
---------------------------------------------------------------
$ wget https://raw.githubusercontent.com/mtpradoc/kafka-spark-demo-pyconsg19/master/iot_demo/kafka_producer_iot.pyTambién contiene el script en Python para hacer la simulación del envio de cada una de las filas como si fuera data en streaming, por ejemplo de un sensor IoT.
Para poder ejecutar este script necesitaremos instalar lo siguiente
a. Python3 — Interprete necesario para poder ejecutar el script
$ sudo yum install python3$ which python3b. Pip3 — Administrador de paquetes de Python necesario para poder instalar las librerías necesarias para ejecutar el script.
$ which pip3 # revisar si se instalo por defecto pip3, sino$ sudo yum install python36-setuptools
$ sudo easy_install-3.6 pipc. Pandas — Librería necesaria para leer los archivos CSV y usar el objeto DataFrame de Pandas que permite el manejo de tablas de una manera sencilla en código
$ sudo pip3 install pandasd. Kafka-Python — Librería necesaria para conectarnos con la máquina Kafka, en especifico con el “Producer” de Kafka usando el puerto 9092.
$ sudo pip3 install kafka-pythonPaso 2 — Broker de Datos en Streaming
Aquí seguiremos los pasos ya mencionados en el articulo de cómo instalar Kafka.
$ cd kafka_2.13-2.5.0# Iniciar Zookeeper
# -----------------$ KAFKA_HEAP_OPTS="-Xmx32M" \
./bin/zookeeper-server-start.sh config/zookeeper.properties > \
/tmp/zookeeper.log &# Iniciar Kafka
# -----------------$ KAFKA_HEAP_OPTS="-Xmx200M" \
./bin/kafka-server-start.sh config/server.properties > \
/tmp/kafka.log 2>&1 &# Listamos los tópicos que hemos creado
# -----------------$ bin/kafka-topics.sh --list --zookeeper localhost:2181Podemos agregar un nuevo tópico o usar uno ya existente. yo usare uno existente “alomarlin”.
Paso 3 — Procesamiento de Datos en Streaming
Seguimos los pasos para instalar Spark en una maquina virtual
En particular en esta maquina usaremos este código.
# Descargamos desde Github el código en PySpark
# ---------------------------------------------$ wget https://raw.githubusercontent.com/mtpradoc/kafka-spark-demo-pyconsg19/master/iot_demo/kafka_producer_iot.pyPara ejecutar este código usaremos un archivo jar en particular. Hay que tener mucho cuidado al momento de usar este archivo pues deberá ser la versión correcta que alinee las versiones de Scala, Spark Streaming SQL y Kafka que se están usando.
Paso 4 — Asegurar la Conectividad de estas Máquinas
Para ello para hacer esto práctico, editamos las reglas en los Security groups de cada máquina virtual que se ha creado. También pudimos poner a las tres maquinas en el mismo Security Group de esta manera aseguramos la conectividad entre estas maquinas a nivel de los puertos de ICMP, 9092 que es el que usan el producer y el consumer (lo hemos dejado en su valor por defecto).
Paso 5 — Procesamiento Extremo a Extremo
Primero — Tenemos la máquina 2 donde tenemos “Kafka” ejecutando y el tópico “alomarlin”.
Segundo — Ejecutamos el script en la máquina virtual 3 “Consumidor de Datos”, para que Spark empiece a escuchar los datos en streaming que vienen de Kafka.
Tercero — Finalmente ejecutamos el script en máquina virtual 1 “Productor de Datos”.
Veremos qué los datos son impresos en consola de la máquina virtual con Spark.
Conclusión
Actualmente existen soluciones administradas en la nube que nos facilitan tener que lidiar con las implementaciones a detalle de cada uno de estos elementos mostrados en la arquitectura. Sin embargo, este esquema es válido donde ya sea por costos o por los volumenes de datos que se manejen o por politicas de las empresas los procesamientos de datos en streaming se tengan que hacer en sitio. Esta vez hemos hecho la implementación de todas estas herramientas en su versión Open Source, pero el mismo procedimiento se usa para versiones comerciales como Cloudera HortonWorks. En el caso de implementaciones en producción, se hablan de clusters de servidores en Kafka, y Spark pues se toma ventaja de la capacidad de procesamiento paralelo que nos ofrecen este tipo de despliegues. En el caso que usemos la nube, solo seria necesario usar Dataproc y correr el archivo en pyspark para ejecutar el procesamiento usando Spark Structured Streaming con data que provenga por ejemplo de PubSub (Kafka para la nube GCP). Y en el caso de AWS, tendriamos que setear EMR para ejecutar nuestro código en pyspark y que la data provenga de Amazon Kinesis (Kafka para AWS). Algo importante a notar aqui es el uso del .jar, a la hora de hacer el spark-submit pues ese archivo al que se hace mención debe ser compatible con la versión de Kafka y Spark, Scala que estemos usando.
