Big Data aplicado: Spark vs Flink

Arquitectura Kappa

La arquitectura Kappa es un patrón de arquitectura de software que utiliza flujos de datos o «streams» en lugar de utilizar datos ya almacenados y persistentes. Las fuentes de datos serán, por tanto, inmutables y continuas, como las entradas generadas en un registro. Estas fuentes de información contrastan con las tradicionales bases de datos utilizadas en los sistemas de análisis por lotes o batch.

Estos «streams» se envían a través de un sistema de procesamiento y se almacenan en sistemas auxiliares para la capa de servicio.

La arquitectura Kappa es una simplificación de la arquitectura Lambda, de la que se diferencia en los siguientes aspectos:

• Todo es un stream (el batch es un caso particular)
• Las fuentes de datos son inmutables (los datos brutos persisten)
• Hay un único marco analítico (beneficioso para el mantenimiento)
• Funcionalidad de reproducción, es decir, reprocesar todos los datos analizados hasta el momento (por ejemplo, al tener un nuevo algoritmo)

En resumen, se elimina la parte del lote, favoreciendo un sistema de procesamiento de datos en movimiento.

Mancharse las manos

Para entender mejor todos estos conceptos teóricos, nada mejor que profundizar en la implementación. Uno de los elementos que queremos poner en práctica con esta prueba de concepto es poder intercambiar entre diferentes motores de procesamiento de forma transparente. Es decir, que sea fácil cambiar entre Spark y Flink partiendo del mismo código Scala. Desgraciadamente, las cosas no son tan sencillas, aunque es posible llegar a una solución parcial común, como se puede ver en las siguientes secciones.

Eventos Netflow

Cada evento Netflow proporciona información sobre una conexión específica establecida en la red. Estos eventos se representan en formato JSON y se ingieren en Kafka. Por simplicidad, sólo se utilizarán conexiones IPv4.
Ejemplo de evento:

{
"eventid":995491697672682,
"dvc_version":"3.2",
"aproto":"failed",
"src_ip4":168099844,
"dst_ip4":4294967295,
"eventtime":1490024708,
"out_packets":1,
"type":"flow",
"dvc_time":1490024739,
"dst_ip":"255.255.255.255",
"src_ip":"10.5.0.4",
"duration":0,
"in_bytes":0,
"conn_state":"new",
"@version":"1",
"out_bytes":157,
"dvc_type":"suricata",
"in_packets":0,
"nproto":"UDP",
"src_port":42126,
"@timestamp":"2017-03-20T15:45:08.000Z",
"dst_port":40237,
"category":"informational"
}

Parte compartida

Como se ha mencionado anteriormente, uno de los objetivos de este artículo es utilizar de forma transparente diferentes entornos de procesamiento de Big Data , es decir, utilizar las mismas funciones de Scala para Spark, Flink o futuros frameworks. Lamentablemente, las peculiaridades de cada framework han hecho que sólo una parte del código Scala implementado pueda ser común para ambos.

El código común incluye los mecanismos para serializar y deserializar los eventos de netflow. Nos apoyamos en la librería json4s de Scala, y definimos una clase case con los campos de interés:

case class Flow_ip4(eventid: String,
dst_ip4: String,
dst_port: String,
duration: String,
in_bytes: String,
in_packets: String,
out_bytes: String,
out_packets: String,
src_ip4: String,
src_port: String) extends Serializable

Además, se define el objeto Flow que implementa esta serialización para poder trabajar cómodamente con diferentes motores de Big Data:

class Flow(trace: String) extends Serializable {implicit val formats =DefaultFormatsvar flow_case : Flow_ip4 = parse(trace).extractdef getFlow() : Flow_ip4 = flow_case}

Otro código compartido por ambos frameworks son las funciones para calcular la distancia euclidiana, obtener las combinaciones de una serie de objetos y generar el Json de resultados:

object utils { def euclideanDistance(xs: List, ys: List) = {
sqrt((xs zip ys).map { case (x,y)=> pow(y - x, 2) }.sum)
} def getCombinations(lists : Iterable) = {
lists.toList.combinations(2).map(x=> (x(0),x(1))).toList
} def getAllValuesFromString(flow_case : Flow_ip4) = flow_case.productIterator.drop(1)
.map(_.asInstanceOf.toDouble).toList def calculateDistances(input: Iterable): List = {
val combinations: List = getCombinations(input)
val distances = combinations.map{
case(f1,f2) => (f1.eventid,f2.eventid,euclideanDistance(getAllValuesFromString(f1),
getAllValuesFromString(f2)))}
distances.sortBy(_._3)
} def generateJson(x: (String, String, Double)): String = {
val obj = Map("eventid_1" -> x._1,
"eventid_2" -> x._2,
"distance" -> x._3,
"timestamp" -> System.currentTimeMillis())
val str_obj = scala.util.parsing.json.JSONObject(obj).toString()
str_obj
}
}

Parte específica de cada Framework

La aplicación de transformaciones y acciones es donde las particularidades de cada framework impiden que sea un código genérico.

Como se ha observado, Flink itera por cada evento, filtrando, serializando, agrupando por IP y tiempo y finalmente aplicando el cálculo de distancia para enviarlo a Kafka a través de un productor.

En Spark, en cambio, como los eventos ya están agrupados por tiempo, sólo tenemos que filtrar, serializar y agrupar por IP. Una vez hecho esto, se aplican las funciones de cálculo de distancia y se puede enviar el resultado a Kafka con las librerías adecuadas.

Para Flink:

val flow = stream
.filter(!_.contains("src_ip6"))
.map(trace =>{
new Flow(trace).getFlow()
})
.keyBy(_.src_ip4)
.timeWindow(Time.seconds(3))
.apply{(
key: String,
window: TimeWindow,
input: Iterable,
out: Collector]) => {
val distances: List = utils.calculateDistances(input)
out.collect(distances)
}
}
.flatMap(x => x)
flow.map( x=>{
val str_obj: String = utils.generateJson(x)
producer.send(new ProducerRecord(topic_out, str_obj))
})

Para Spark:

val flow = stream.map(_.value())
.filter(!_.contains("src_ip6"))
.map(record => {
implicit val formats = DefaultFormats
val flow = new Flow(record).getFlow()
(flow.src_ip4,flow)
})
.groupByKey()flow.foreachRDD(rdd => {
rdd.foreach{ case (key, input ) => {
val distances = utils.calculateDistances(input)
distances.map(x => {val str_obj: String = utils.generateJson(x)
producer.send(new ProducerRecord(topic_out, str_obj))
})
}}
})

En el tema de salida de Kafka se almacenan los IDs de cada evento, sus timestamps de ingesta, el resultado del cálculo de distancia y la hora a la que se completó el cálculo. A continuación se muestra un ejemplo de rastreo:

{
"eventid_1":151852746453199,
"ts_flow1":1495466510792,
"eventid_2":1039884491535740,
"ts_flow2":1495466511125,
"distance":12322.94295207115,
"ts_output":1495466520212
}

Resultados

Hemos realizado dos tipos diferentes de pruebas: múltiples eventos dentro de una misma ventana, y múltiples eventos en diferentes ventanas, donde un fenómeno hace que el algoritmo combinatorio sea aparentemente estocástico y ofrezca diferentes resultados. Esto ocurre debido a que los datos se ingieren de forma desordenada, a medida que llegan al sistema, sin tener en cuenta el tiempo del evento, y se hacen agrupaciones en base a él, lo que hará que existan diferentes combinaciones.

La siguiente presentación de Apache Beam explica con más detalle la diferencia entre el tiempo de procesamiento y el tiempo del evento: https://goo.gl/h5D1yR