Primeros pasos con Apache Spark 2

Posted on agosto 10, 2016 por admin

Hace pocos días salió la esperada versión 2 de Apache Spark y como algunos de ustedes saben es un framework que ahora mismo atrae mucho mi atención y como no pudo ser de otra forma hice un pequeño proyecto donde quiero ir colocando ejemplos sencillos de Spark con las nuevas (y no tan nuevas) cosas de Spark.

Para empezar comentarles que yo todavía no he utilizado sbt sino por el contrario uso maven como herramienta de construcción de proyectos. He aquí los primeros cambios necesarios para trabajar con spark 2, las dependencias correspondientes a la versión (indicadas en el pom.xml).

   <dependencies>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.10</artifactId>
      <version>2.0.0</version>
    </dependency>

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.10</artifactId>
      <version>2.0.0</version>
    </dependency>
    
   ...

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.10</artifactId>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-sql_2.10</artifactId>

</dependency>

...

Entre los nuevos cambios de spark está que el punto de entrada para los programas spark ya no serán el hiveContext o sqlContext sino que han sido subsumidas en una clase llamada SparkSession. Las clases HiveContext y SQLContext se han mantenido para proporcionar retrocompatibilidad. Ejemplo

// En spark v1.6.2 &lt;= lo hubieramos hecho asi
  //  val conf = new SparkConf().setMaster("local").setAppName("Simple Application")
  //  val sc = new SparkContext(conf)
  //  val sqlContext = new org.apache.spark.sql.SQLContext(sc)
  val sparkSession = SparkSession.builder.
    master("local")
    .appName("Simple Application")
    .getOrCreate()

// En spark v1.6.2 <= lo hubieramos hecho asi

// val conf = new SparkConf().setMaster("local").setAppName("Simple Application")

// val sc = new SparkContext(conf)

// val sqlContext = new org.apache.spark.sql.SQLContext(sc)

val sparkSession = SparkSession.builder.

master("local")

.appName("Simple Application")

.getOrCreate()

Con el SparkSession haremos lo mismo que hacíamos con sqlContext por ejemplo obtener un Dataset

  //obtenemos el dataset de tipo People
  val ds = sparkSession.read.json("src/main/resources/people.json").as[People]

1 2	//obtenemos el dataset de tipo People val ds = sparkSession.read.json("src/main/resources/people.json").as[People]

O por el contrario obtener un DataFrame

val df = sparkSession.read.format("com.databricks.spark.csv")
    .format("com.databricks.spark.csv")
    .option("header", "true") // Use first line of all files as header
    .option("inferSchema", "true") // Automatically infer data types
    .load("src/main/resources/datos.csv")

val df = sparkSession.read.format("com.databricks.spark.csv")

.format("com.databricks.spark.csv")

.option("header", "true") // Use first line of all files as header

.option("inferSchema", "true") // Automatically infer data types

.load("src/main/resources/datos.csv")

Otro punto importante ha sido la unificación de las clases Dataset y DataFrame (para Java y Scala) a partir de la versión 2.¿Qué significa esto? pues sencillamente que ahora solo existirá la clase Dataset, pero proporcionará la misma funcionalidad que nos daba la clase DataFrame, de hecho basta con comparar la API en la versión 1.6.2 y 2.0.0 y ver como los métodos de la clase DataFrame están ahora incluidos en la clase Dataset.

Dataset y Dataframe en Spark 2

Aquellos interesados en leer más acerca de Dataset y Dataframe visitar este link

Estos no son los únicos cambios en Spark, de hechos son muchos más, que se corresponden a optimizaciones a nivel de compilación y ejecución así como también a un nuevo parseador SQL, para leer mas acerca de lo nuevo en Spark 2 clic aquí.

Aqui les dejo en enlace al proyecto donde ire añadiendo clases y seguiré probando mas cosas nuevas de Spark.

Estadística simple con Spark

Posted on julio 11, 2015 por admin

Hace unos 20 días lei un articulo titulado «Simple Data Analysis Using Spark» (no publico el enlace ya que el mismo desapareció de dzone de la zona de Big Data y el mismo blog ya no existe, corroborarlo buscando por Internet), lo interesante de este articulo es que el autor hacía cálculos de estadística simple, es decir, calculaba, media, máximo y mínimo, y fue casualidad que justo en ese momento acaba de leer acerca de las operaciones numéricas de los RDD, es esta la razón que me movió a querer hacer un ejercicio similar, con estadística simple pero quería hacerlo con una información que yo considerara valiosa.

Fue entonces que me decidí a buscar alguna fuente de open data española, ya que quería arrojar u obtener datos reales, o no se quizás quería sencillamente obtener algo significativo que se acercara a una tarea real, fue así que llegue al Portal de datos abiertos de la comunidad de Madrid y después de revisar el catalogo me decidí por el padrón municipal, un fichero csv de 22 MB.

He aquí el código:

package com.josedeveloper

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._

case class Padron(codDistrito: String, descDistrito: String, codDistBarrio: String,
             descBarrio: String, codBarrio: String, codDistSeccion: String, spainMen: Integer,
             spainWomen: Integer, foreignerMen: Integer, foreignerWomen: Integer) {

  def numberOfPeople() : Int = {
    return spainMen + spainWomen + foreignerMen + foreignerWomen;
  }

}

object ScalaApp extends App {

  val sc = new SparkContext("local", "Simple Application", "$SPARK_HOME"
    , List("target/SimpleApp-0.1.jar"))

  val file = sc.textFile("Rango_Edades_Seccion_201506.csv")

  val data = file.map(line =&gt; line.split(";").map(_.trim))
      //the csv header is excluded
      .mapPartitionsWithIndex((idx, iter) =&gt; if (idx == 0) iter.drop(1) else iter).persist()

  //Map to Padron objects
  val dataMapped = data.map(line =&gt; (line(1), new Padron(line(0), line(1),
    line(2), line(3), line(4), line(5),
    getIntValue(line(8)),
    getIntValue(line(9)),
    getIntValue(line(10)),
    getIntValue(line(11)))))

  //Groupped by district
  val grouppedData = dataMapped.reduceByKey((x: Padron, y: Padron) =&gt; new Padron(x.codDistrito, x.descDistrito,
    x.codDistBarrio, x.descBarrio, x.codBarrio, x.codDistSeccion, x.spainMen + y.spainMen,
    x.spainWomen + y.spainWomen, x.foreignerMen + y.foreignerMen,
    x.foreignerWomen + y.foreignerWomen))

  //sorted list total spanish men by district
  grouppedData.collect().sortBy(_._2.spainMen)
    .foreach(x =&gt; println(x._2.descDistrito + "-&gt;" + x._2.spainMen))

  //statistics and numerics sparks RDD values
  val spainMaleValeByDistrict = grouppedData.map(tupla =&gt; tupla._2.spainMen.doubleValue()).cache()
  val media = spainMaleValeByDistrict.mean()
  val stddev = spainMaleValeByDistrict.stdev()
  val max = spainMaleValeByDistrict.max()
  val min = spainMaleValeByDistrict.min()

  println("Media de Españoles varones por distrito: " + media.toInt)
  println("Desviación estandar de españoles varones por distrito: " + stddev.toInt)
  println("Num maximo de españoles varones en un distrito: " + max.toInt)
  println("Num minimo de españoles varones en un distrito: " + min.toInt)


  //sorted list total number of people by district
  grouppedData.collect().sortBy(_._2.numberOfPeople)
    .foreach(x =&gt; println(x._2.descDistrito + "-&gt;" + x._2.numberOfPeople))

  val numberOfPeopleByDistrict = grouppedData.map(tupla =&gt; tupla._2.numberOfPeople.doubleValue).cache()
  val media2 = numberOfPeopleByDistrict.mean()
  val stddev2 = numberOfPeopleByDistrict.stdev()
  val max2 = numberOfPeopleByDistrict.max()
  val min2 = numberOfPeopleByDistrict.min()

  println("Media de personas por distrito: " + media2.toInt)
  println("Desviación estandar de personas por distrito: " + stddev2.toInt)
  println("Num maximo de personas en un distrito: " + max2.toInt)
  println("Num minimo de personas en un distrito: " + min2.toInt)

  def getIntValue(s:String) : Integer = {
    val value = s.substring(1, s.length - 1)
    if (value.isEmpty)
      return 0
    return Integer.parseInt(value)
  }

}

package com.josedeveloper

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

case class Padron(codDistrito: String, descDistrito: String, codDistBarrio: String,

descBarrio: String, codBarrio: String, codDistSeccion: String, spainMen: Integer,

spainWomen: Integer, foreignerMen: Integer, foreignerWomen: Integer) {

def numberOfPeople() : Int = {

return spainMen + spainWomen + foreignerMen + foreignerWomen;

}

object ScalaApp extends App {

val sc = new SparkContext("local", "Simple Application", "$SPARK_HOME"

, List("target/SimpleApp-0.1.jar"))

val file = sc.textFile("Rango_Edades_Seccion_201506.csv")

val data = file.map(line => line.split(";").map(_.trim))

//the csv header is excluded

.mapPartitionsWithIndex((idx, iter) => if (idx == 0) iter.drop(1) else iter).persist()

//Map to Padron objects

val dataMapped = data.map(line => (line(1), new Padron(line(0), line(1),

line(2), line(3), line(4), line(5),

getIntValue(line(8)),

getIntValue(line(9)),

getIntValue(line(10)),

getIntValue(line(11)))))

//Groupped by district

val grouppedData = dataMapped.reduceByKey((x: Padron, y: Padron) => new Padron(x.codDistrito, x.descDistrito,

x.codDistBarrio, x.descBarrio, x.codBarrio, x.codDistSeccion, x.spainMen + y.spainMen,

x.spainWomen + y.spainWomen, x.foreignerMen + y.foreignerMen,

x.foreignerWomen + y.foreignerWomen))

//sorted list total spanish men by district

grouppedData.collect().sortBy(_._2.spainMen)

.foreach(x => println(x._2.descDistrito + "->" + x._2.spainMen))

//statistics and numerics sparks RDD values

val spainMaleValeByDistrict = grouppedData.map(tupla => tupla._2.spainMen.doubleValue()).cache()

val media = spainMaleValeByDistrict.mean()

val stddev = spainMaleValeByDistrict.stdev()

val max = spainMaleValeByDistrict.max()

val min = spainMaleValeByDistrict.min()

println("Media de Españoles varones por distrito: " + media.toInt)

println("Desviación estandar de españoles varones por distrito: " + stddev.toInt)

println("Num maximo de españoles varones en un distrito: " + max.toInt)

println("Num minimo de españoles varones en un distrito: " + min.toInt)

//sorted list total number of people by district

grouppedData.collect().sortBy(_._2.numberOfPeople)

.foreach(x => println(x._2.descDistrito + "->" + x._2.numberOfPeople))

val numberOfPeopleByDistrict = grouppedData.map(tupla => tupla._2.numberOfPeople.doubleValue).cache()

val media2 = numberOfPeopleByDistrict.mean()

val stddev2 = numberOfPeopleByDistrict.stdev()

val max2 = numberOfPeopleByDistrict.max()

val min2 = numberOfPeopleByDistrict.min()

println("Media de personas por distrito: " + media2.toInt)

println("Desviación estandar de personas por distrito: " + stddev2.toInt)

println("Num maximo de personas en un distrito: " + max2.toInt)

println("Num minimo de personas en un distrito: " + min2.toInt)

def getIntValue(s:String) : Integer = {

val value = s.substring(1, s.length - 1)

if (value.isEmpty)

return 0

return Integer.parseInt(value)

}

Hay una cosa que quiero resaltar, Lo simple y corto del código en Scala, tanto para la definición de la clase Padron como en las transformaciones hechas en los RDD, ya Scala esta en mi lista de cosas por aprender porque se que con Java hubiese sido el doble de código.

Por último les dejo parte de la información obtenida, la lista ordenada de población por distrito:

BARAJAS: 46166
VICALVARO: 69709
MORATALAZ: 94785
VILLA DE VALLECAS: 101011
MONCLOA-ARAVACA: 116581
RETIRO: 118390
CENTRO: 131805
USERA: 133579
CHAMBERI: 137454
VILLAVERDE: 141457
CHAMARTIN: 142334
SALAMANCA: 143123
ARGANZUELA: 151061
TETUAN: 152142
SAN BLAS-CANILLEJAS: 153303
HORTALEZA: 176320
CIUDAD LINEAL: 212626
PUENTE DE VALLECAS: 227183
LATINA: 234842
FUENCARRAL-EL PARDO: 234883
CARABANCHEL: 242032

¿Te ha parecido interesante? ¿Qué le agregarías o quitarías al código?

Si quieres acceder al GitHub donde he colgado el código pincha aquí

Primer ejemplo con apache spark

Posted on junio 12, 2015 por admin

Hace algunas semanas atrás que empezamos a trabajar con apache spark en el máster, brevemente les contare mis impresiones desde mi punto de vista como principiante.

Apache spark me gusto, ¿por qué? porque se puede programar en Scala (además de Python y Java), la API de Scala lo simplifica mucho y la cantidad de código a teclear es (considerablemente) menor a la necesaria para hacer la misma tarea en Java, aunque aquí debo hacer un paréntesis, ya que con la entrada de Java 8 y las lambda expresión la cantidad de código será menor pero insisto la API de Scala a mi modo de ver lo hace mas sencillo.

Nos permite hacer operaciones con mucha data (quizás no Big Data, es decir no hablamos de TeraBytes) sin necesidad de usar Hadoop, me gustaría hacer hincapié en esto ya que muchos piensan que Big Data es Hadoop y NO, no es así, mi concepto de Big Data es un poco mas amplio y contempla mas cosas, aquí me refiero específicamente a la API MapReduce de Hadoop la cual es mas complicada de usar que la de Spark, no obstante si necesitamos trabajar con una cantidad de datos considerable que requiera del uso de HDFS, podemos desde Spark acceder a este sistema de archivo u algún otro como por ejemplo Amazon S3.

Por último decirles que Spark es mas rapido, esto se debe a su arquitectura, ya que este trabaja en memoria (todo en memoria si tiene la capacidad de cargar todos los datos por completo) y pues si llegamos a hacer varias tareas (lo que técnicamente sería reutilizar los RDD ya guardados en memoria) con los datos en memoria es cuando mas jugo se le sacará a Spark.

Sinceramente si alguien discrepa de mi o coincide me entusiasmaría mucho que comparta su opinión ya que todo este mundo que envuelve el Big Data me tiene super enganchado.

Ahora bien vamos con los ejemplos, estas formaban parte de la primera tarea de Spark que tuvimos que realizar, y consiste en 2 ejemplos:

Dado un fichero de texto en formato csv con información de medallistas olímpicos, obtener el numero de medallas por edad.
Dado el mismo fichero de texto anterior, y un esquema de puntuación por tipo de medalla obtener un ranking de medallistas olímpicos.

He aqui el código:

package org.dummy

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._

object ScalaApp extends App {
  val logFile = "OlympicAthletes.csv"
  val sc = new SparkContext("local", "Simple", "$SPARK_HOME"
    , List("target/spark-ejercicio-1.0.jar"))

  //Todo el fichero csv es cargado en este RDD
  val file = sc.textFile(logFile)

  //linea a linea se aplica un split o separacion utilizando la ","
  //eso nos arrojara un arreglo de palabras (o numeros)
  //y construimos una instancia del tipo OlympicMedalRecords
  val olympicMedalRecordsRDD = file.map(x =&gt; {
    val arr = x.split(",")
    new OlympicMedalRecords(arr(0), Integer.parseInt(arr(1)), arr(2)
      , Integer.parseInt(arr(3)), arr(5), Integer.parseInt(arr(6)),
      Integer.parseInt(arr(7)), Integer.parseInt(arr(8)))
  }
  )


  //Mis ejercicios
  //Ejercicio 1
  //mapeo del tipo (edad, num de medallas), val medal es un RDD
  val medal = olympicMedalRecordsRDD.map(record =&gt; (record.getAge, record.getGoldMedals + record.getSilverMedals + record.getBronzeMedals))

  //creamos un nuevo RDD el cual manejara la suma de todas las madellas por edad
  val sum = medal.reduceByKey((acc: Int,value: Int)=&gt;acc+value)
  println("Lista de Medallas ordenado por Edad")

  //ordenamos, juntamos y luego recorremos para mostrar todos los valores
  sum.sortBy(_._1).collect.foreach(println)


  //Ejercicio 2 hacer un ranking por atleta
  // ya que oro = 3 ptos, plata = 2ptos, bronce = 1pto
  //sportGuy es otro RDD y aqui hacemos un mapeo del tipo (nombre_atleta, puntos_por_medalla)
  val sportGuy = olympicMedalRecordsRDD.map(record =&gt; (record.getName, 3*record.getGoldMedals + 2*record.getSilverMedals + record.getBronzeMedals))

  //ranking es a su vez otro RDD  y de igual manera como se hizo en el ejemplo anterior se agrupa por nombre de atleta
  // y también utilizamos una variable acumuladora acc donde vamos sumando los ptos de un determinado atleta
  val ranking = sportGuy.reduceByKey((acc,value)=&gt;acc+value)
  println("Ranking de deportistas")
  // ordenamos en sentido inverso y mostramos  los atletas con mas ptos
  ranking.sortBy(_._2, false).collect.foreach(println)
}

package org.dummy

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

object ScalaApp extends App {

val logFile = "OlympicAthletes.csv"

val sc = new SparkContext("local", "Simple", "$SPARK_HOME"

, List("target/spark-ejercicio-1.0.jar"))

//Todo el fichero csv es cargado en este RDD

val file = sc.textFile(logFile)

//linea a linea se aplica un split o separacion utilizando la ","

//eso nos arrojara un arreglo de palabras (o numeros)

//y construimos una instancia del tipo OlympicMedalRecords

val olympicMedalRecordsRDD = file.map(x => {

val arr = x.split(",")

new OlympicMedalRecords(arr(0), Integer.parseInt(arr(1)), arr(2)

, Integer.parseInt(arr(3)), arr(5), Integer.parseInt(arr(6)),

Integer.parseInt(arr(7)), Integer.parseInt(arr(8)))

}

)

//Mis ejercicios

//Ejercicio 1

//mapeo del tipo (edad, num de medallas), val medal es un RDD

val medal = olympicMedalRecordsRDD.map(record => (record.getAge, record.getGoldMedals + record.getSilverMedals + record.getBronzeMedals))

//creamos un nuevo RDD el cual manejara la suma de todas las madellas por edad

val sum = medal.reduceByKey((acc: Int,value: Int)=>acc+value)

println("Lista de Medallas ordenado por Edad")

//ordenamos, juntamos y luego recorremos para mostrar todos los valores

sum.sortBy(_._1).collect.foreach(println)

//Ejercicio 2 hacer un ranking por atleta

// ya que oro = 3 ptos, plata = 2ptos, bronce = 1pto

//sportGuy es otro RDD y aqui hacemos un mapeo del tipo (nombre_atleta, puntos_por_medalla)

val sportGuy = olympicMedalRecordsRDD.map(record => (record.getName, 3*record.getGoldMedals + 2*record.getSilverMedals + record.getBronzeMedals))

//ranking es a su vez otro RDD y de igual manera como se hizo en el ejemplo anterior se agrupa por nombre de atleta

// y también utilizamos una variable acumuladora acc donde vamos sumando los ptos de un determinado atleta

val ranking = sportGuy.reduceByKey((acc,value)=>acc+value)

println("Ranking de deportistas")

// ordenamos en sentido inverso y mostramos los atletas con mas ptos

ranking.sortBy(_._2, false).collect.foreach(println)

}

Una última cosa que es bueno saber para aquellos que como yo están empezando con Spark y sus RDD, Los RDD tienen 2 tipos de operaciones: transformaciones y acciones, las primeras son operaciones que como resultado nos devuelven otro RDD y las acciones son aquellas que hacen «algo» sobre los datos como por ejemplo el foreach de mi código, utilizado para mostrar por pantalla los valores finales, además los RDD (como hasta ahora lo entiendo) son las instrucciones a realizar sobre los datos, mas no almacenan la información, además estás instrucciones no se realizan de inmediato, sino que al contrario estos tienen un funcionamiento lazy, esto significa que se ejecutan cada vez que se realiza una operación de tipo acción.

Otra sugerencia, no duden en cacharrear con la consola de spark, de hecho el primer ejercicio lo llegue a hacer por la consola (sin los JavaBeans) y el practicar por aquí si que ayuda a conocer los operaciones a utilizar sobre los RDD.

Aquellos interesados en descargar el código, les dejo la dirección del ejercicio en mi repositorio github

El esqueleto del ejercicio esta en el repositorio del que fue nuestro profesor y pueden acceder a el haciendo clic aquí.

Cualquier material, idea o contenido que quieres compartir, no dudes en hacerlo, ya que bien recibido será. Ahora si para cerrar les dejo el enlace a un libro que hasta ahora me esta pareciendo muy bueno para aprender Spark (además con ejemplos elaborados en Java, Scala y Python).

El Blog de Jose

Blog de tecnología, software y programación

Archivo de la etiqueta: spark

Primeros pasos con Apache Spark 2

Estadística simple con Spark

Primer ejemplo con apache spark

Uso de cookies