Apache Spark RDDs VS Dataframes

Te explicamos las diferencias entre los RRDs y los Dataframes de Apache Spark.

Diferentes APIs de Spark

Spark dispone de 3 APIs para el desarrollo de las aplicaciones:

• RDD, que surgió en la versión 1.0.
• DataFrame, que surgió en la versión 1.3.
• DataSet, que surgió en la versión 1.6.

Hay que tener en cuenta que a partir de Spark 2.0, el API de DataFrame se engloba dentro de DataSet, y se recomienda basar el desarrollo en estas dos APIs.

El tipo RDD no es que haya desaparecido, ya que está presente dentro DataSet y es muy importante de cara al rendimiento de Spark, pero lo vamos a utilizar para implementar tareas a bajo nivel, y en la práctica lo que vamos a usar cuando estemos con el Spark SQL será DataSet o DataFrame.

DataFrame no es más que una alias de DataSet[Row], siendo Row un objeto genérico en la máquina virtual de Java.

Uso de DataFrame y DataSet

El uso de DataFrame y DataSet es muy similar, simplemente tendremos un DataSet cuando queramos tener un tipo customizado de filas en la máquina virtual Java.

De esta forma el compilador puede detectar errores de sintaxis sobre algún campo. Por ejemplo, si hacemos un select sobre una columna insistente, con DataFrame solo es posible ver el error al correr el programa, mientras que con DataSet es posible que el compilador detecte ese error sin completar la ejecución.

Preferentemente vamos a usar el tipo de DataFrame, que es el que se tiene por defecto al ejecutar un reader, pero cuando queramos especificar el tipo de datos en el esquema vamos a meter un DataSet, y para especificar los tipos se utiliza una case class.

En filtros esto permite que los fallos sintácticos sean detectables. Respecto a RDD, en general no se debe usar. Aquí podéis obtener más información al respecto:

https://dzone.com/articles/apache-spark-3-reasons-why-you-should-not-use-rdds https://databricks.com/blog/2016/07/14/a-tale-of-three-apache-spark-apis-rdds-dataframes-and- datasets.html

Ejemplo de uso de case class

Imaginemos que tenemos un empleado con el nombre, sexo, edad y sueldo, podríamos definir un DataSet leyendo un CSV, definiendo este esquema a partir de las case class y cargando el fichero correspondiente.

case class Empleado(Nombre: String,Sexo: String,Edad:Integer,Sueldo: Integer)
import org.apache.spark.sql.Encoders
val esquemaDeLaClase = Encoders.product[Empleado].schema
val miDataset=spark.read.format(“csv”).

schema(esquemaDeLaClase).option(“header”,”true”).
load(“C:/Users/psz/Desktop/claseIntroSpark/video/Sueldos.csv”).as[Empleado]
miDataSet.show

Respecto a la diferencia en la práctica entre tener un DatSet y un DataFrame es la forma de utilizar el filtro:

• En caso de DataFrame se utiliza el comando col(nombre) o el comando $”nombre”, siendo nombre el nombre de la columna.
• En caso del DatSet permite, además, acceder al campo de cada fila con .nombre, lo cual permite al compilador saber si nos estamos equivocando en el nombre.

Uso de RDD

RDD se utiliza para definición manual de conjuntos, mediante el método parallelize.

Por ejemplo, si tenemos POJOs de Java, después haríamos una serie de operaciones map y finalmente, antes de calcular gradaciones lo convertiríamos a DataFrame para que mejore el rendimiento.

Ejemplo

Vamos a ver un ejemplo de cómo definir una secuencia con una clase de Java.

En la misma importamos la clase Mensaje, definimos dos mensajes y una secuencia con estas dos variables de Java:

import paqmensajeria.Mensaje
val m1=new Mensaje(“Juan”,”Ortega Varona”,1, “Mario”, “Campo Bueno”,2, “Solicitud Admisión”,
“Deseo unirme al grupo G0”)
val m2=new Mensaje(“Mario”, “Campo Bueno”,2, “Juan”, “Ortega Varona”,1, “Admitido”,
“!Bienvenido a G0!”)
val listaMensajes=Seq(m1,m2)

Una vez que hemos hecho esta secuencia, el método parallelize lo que hace es convertir la secuencia en un RDD:

val RDD=sc.parallelize(listaMensajes)

Luego podemos hacer una serie de transformaciones con Map, para dejar la estructura más plana, algo más parecido a una tabla:

	
var rDD2=miRDD.map(x=>(x.getRemitente.getId,x.getDestinatario.getId,x.getTexto))
rDD2.foreach(println)

Finalmente, para que el rendimiento sea el adecuado, convertimos con DataFrama, con el que tenemos dos opciones: podemos decir únicamente los nombres de las columnas, o especificar una case class para especificar un esquema:

// Sin esquema
val dfConEsquema =
spark.createDataFrame(rDD2).toDF(“IdRemitente”, “IdDestinatario”, “Texto”)
dfConEsquema.show
// Con Esquema
case class Fila(id1: String, id2: String, texto: String)
val rDDcaseClass=miRDD.map(x=>Fila(
x.getRemitente.getId.toString,
x.getDestinatario.getID.toString,x.getTexto))

val df3 = Spark.createDataFrame(rDDcaseClass).
toDF(“IdRemitente”, “IdDestinatario”, “Texto”)

Aprende a programar scripts de procesamiento de datos capaces de ejecutarse de forma clusterizada mediante el framework Apache Spark.

Recuerda que puedes comenzar este curso con tu suscripción de OpenWebinars. Si todavía no estás suscrito, aprovecha para hacerlo ahora.