Concepto de ventana en Apache Spark Streaming

Hemos hecho un ejemplo con Apache Spark Streaming. Pero ahora, consideremos la idea de Windows. En Spark Streaming, tenemos pequeños lotes de datos que ingresan en un rango de tiempo o ventana de tiempo.

Spark hace un lote de los datos entrantes de acuerdo con el intervalo de tiempo, pero a veces debemos recordar cosas del pasado. Por ejemplo podemos querer mantener un promedio de treinta segundos para datos de entrada, pero queremos resultados cada cinco segundos. En este caso, desearía un intervalo de lote de cinco segundos, pero una longitud de ventana de treinta segundos. Spark proporciona varios métodos para realizar este tipo de cálculos.

La solución que tenemos tenemos que usar son funciones de ventana.

Windows nos permite tomar un primer lote y luego un segundo lote y luego un tercer lote y luego crear una ventana de todos los lotes en función del intervalo de tiempo especificado. De esta forma, siempre podemos tener el nuevo RDD y también el historial de los RDD que existieron en la ventana.

La función de ventana más simple es una ventana, que le permite crear una nueva DStream, calculada aplicando los parámetros de ventana al viejo DStream. Puede usar cualquiera de las operaciones de DStream en la nueva transmisión, para que tenga toda la flexibilidad que pueda desear.

Por ejemplo, desea PUBLICAR a todos los usuarios activos de los últimos cinco segundos, pero desea actualizar los resultados cada segundo.

sc = SparkContext(appName="ActiveUsers")
ssc = StreamingContext(sc, 1)
activeUsers = [
    ["Alice", "Bob"],
    ["Bob"],
    ["Carlos", "Dan"],
    ["Carlos", "Dan", "Erin"],
    ["Carlos", "Frank"],
]
rddQueue = []
for datum in activeUsers:
    rddQueue += [ssc.sparkContext.parallelize(datum)]
inputStream = ssc.queueStream(rddQueue)
inputStream.window(5, 1)\
    .map(lambda x: set([x]))\
    .reduce(lambda x, y: x.union(y))\
    .pprint()
ssc.start()
sleep(5)
ssc.stop(stopSparkContext=True, stopGraceFully=True)

Este ejemplo imprime todos los usuarios activos de los últimos cinco segundos, pero lo imprime cada segundo. No es necesario realizar un seguimiento manual del estado, ya que la función de ventana conserva los datos antiguos durante otros cuatro intervalos. La función de ventana le permite especificar la duración de la ventana y la duración de la diapositiva, o la frecuencia con la que desea que se calcule una nueva ventana.

Un ejemplo utilizando Apache Spark Streaming

Lo que vamos a hacer es un proceso que escuche lo que escribimos con "nc" en el puerto 9999. Y luego vamos a contar, es decir vamos a contar las palabras.

Iniciamos spark-shell :

./spark-shell

Luego debemos importar todos los objetos para hacer streaming:

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._

//Ahora hacemos el contexto de streaming, utilizando el contexto de mi aplicación. A la vez //especificamos el periodo de tiempo que se debe procesar los datos:

val ssc = new StreamingContext(sc, Seconds(10))

// Creamos el DStream para localhost:9999
val lines = ssc.socketTextStream("localhost", 9999)

// indicamos las tareas a realizar con los datos.
val words = lines.flatMap(_.split(" "))
val pairs = words.map(word => (word, 1))

val wordCounts = pairs.reduceByKey(_ + _)

// Imprima los primeros elementos de cada RDD generado en este DStream para la consola

wordCounts.print()

ssc.start()             // comenzamos
ssc.awaitTermination()  // Espere a que termine el cálculo

Y listo, ahora si vamos a otra consola y escribimos :

nc -lk 9999

Este comando va a escribir lo escribimos en el puerto 9999 y nuestra aplicación va a leerlo y contarlo.

Y eso es todo!!!

Spark Streaming API

Spark Streaming viene con un conjunto de métodos que facilitan el procesamiento de datos. Hay operaciones que son compatibles con RDD como : map , flatMap ,filter , count , reduce , groupByKey , reduceByKey , sortByKey y join. A la vez provee un conjunto de métodos basados en ventanas y operaciones stateful como :  window , countByWindow , reduceByWindow , countByValueAndWindow , reduceByKeyAndWindow y updateStateByKey.

Spark streaming soporta a scala, java y python.

Vamos a ver los pasos típicos para hacer una aplicación con Spark Streaming:

El primer paso es generar el contexto el cual tiene 2 parámetros en el constructor. Uno es el contexto de spark (que lo tenemos) y el otro es el sliding-interval time, este intervalo es el intervalo en el cual los datos van a ser actualizados, es decir es estamos definiendo en que intervalo se van a buscar los datos. Una vez que se inicializa el contexto, no se pueden definir ni agregar nuevos cálculos al contexto existente. Además, solo un objeto StreamingContext puede estar activo en cualquier momento.

Luego tenemos que definir el origen de datos de entrada esto lo hacemos definiendo un DStream de entrada.

Teniendo el origen de datos definimos los cálculos que se realizarán en la entrada DStream utilizando la API de transformaciones Spark Streaming.

Después de definir la lógica de cómputo de transmisión, podemos comenzar a recibir los datos y procesarlos usando el método de inicio en el objeto StreamingContext.

Finalmente, esperamos que el proceso de transmisión de datos se detenga utilizando el método awaitTermination del objeto StreamingContext.

StreamingContext

Al igual que SparkContext lo ultilizamos para trabajar con spark,  para trabajar con Streaming en spark tenemos StreamingContext que es el principal punto de entrada para todas las funciones de transmisión.

Usando este contexto podemos crear un DStream que representa datos de streaming desde un destino TCP, especificando el hostname y el puerto. Por ejemplo si queremos utilizar una herramienta como Ncat para probar Spark Streaming, recibiríamos una secuencia de datos de la máquina donde se está ejecutando Ncat (por ejemplo, localhost) y el número de puerto de 9999.

Tengamos en cuanta que Spark funciona de modo perezoso, de tal manera cuando configuramos el cálculo que realizará cuando se inicie, y no cuando se vaya configurando. 

Para iniciar el procesamiento (después de que se hayan configurado todas las transformaciones) llamamos al método start() para iniciar el cómputo y al método awaitTermination() para esperar a que finalice.

Dstream

Dstream es la abstracción básica en Spark Streaming y representa una stream continuo de datos.

DStream se puede crear a partir de flujos de datos de entrada procedentes de fuentes como Kafka, Flume y Kinesis, o aplicando operaciones en otros DStream. Internamente, un DStream se representa como una secuencia de objetos RDD.

Similar a RDD el DStream soporta:

• map 
• flatMap 
• filter 
• count 
• reduce 
• countByValue 
• reduceByKey 
• join 
• updateStateByKey 

Apache Spark Streaming

Con el procesamiento de datos en streaming podemos analizar los datos en tiempo real, lo cual es muy útil en muchas situaciones. Apache Spark provee una librería para procesar streaming, que como no era necesarío un nombre difícil o original lo bautizaron Apache Spark Streaming.

El procesamiento de streaming se utiliza para analizar información que viene de sensores, iot, busquedas, logs de paginas web, el trafico a servidores, etc. Ejemplos de su uso podrian ser: monitorización de servidores o del comportamiento de usuarios en paginas web, etc.

Spark Streaming es una extensión de Apache Spark API y como ya digimos Apache Spark Streaming hace facil la creación de procesos tolerante a fallos para procesar información en tiempo real.

Desde la versión 2.0, Spark streaming soporta una nueva librería de streaming llamada 
Structured Streaming, el cual ofrece un procesamiento de streaming escalable y tolerante a fallos basado en Spark SQL. Podemos utilizar dataset and la Api de dataframe en Scala, Java, Python o R para escribir agregaciones de streamings. Structured Streaming provee un camino para el procesamiento sin que tengamos que razonar mucho sobre el procesamiento.

Spark Streaming funciona dividiendo la transmisión en pequeñas porciones de datos  (llamados micro-batches) en un intervalo predefinido (N segundos) y luego convierte cada micro-batch en un RDD. Nosotros podemos procesar estos RDD con las operaciones : map , reduce , reduceByKey , join , y window. Los resultados de estas operaciones RDD se devuelven en lotes. Por lo general, almacenamos estos resultados en un almacén de datos para un análisis posterior, para generar informes, o para enviar alertas basadas en eventos.

Es importante configurar correctamente el intervalo de tiempo para Spark Streaming, en función del caso de uso y los requisitos de procesamiento de datos. Si el valor de N es demasiado bajo, entonces los micropaquetes no tendrán suficientes datos para dar resultados significativos durante el análisis.

Los datos de streaming pueden ser procesados desde diferentes fuentes :

• Kafka,
• Flume,
• Twitter,
•  ZeroMQ,
•  Amazon’s Kinesis, and
•  TCP sockets

Otra ventaja de utilizar Apache Spark es que podemos combinar procesamiento batch con procesamiento streaming. También podemos utilizar otros subproyecto somo Spark mlib o graphX.