Los acumuladores son variables compartidas entre ejecutores que normalmente se utilizan para agregar contadores a su programa Spark.
Spark admite de forma predeterminada la creación de acumuladores de cualquier tipo numérico y proporciona la capacidad de agregar tipos de acumuladores personalizados. Los acumuladores se pueden usar para implementar contadores (como en MapReduce) o sumas. Spark admite de forma nativa acumuladores de tipos numéricos y los programadores pueden agregar compatibilidad con nuevos tipos.
Como usuario podemos crear dos tipos de acumuladores en Spark: acumuladores sin nombre y acumuladores con nombre. Los acumuladores con nombre se mostrarán en la UI web para la etapa que modifica ese acumulador. Spark muestra el valor de cada acumulador modificado por una tarea en la tabla de tareas(Tasks en inglés). La siguiente imagen muestra donde podemos localizar esta información dentro de la UI. Para este caso se ha utilizado el acumulador longAcc.
Acumulador long
Para crear un acumulador log debemos utilizar el método longAccumulator() que está disponible desde el sparkContext.
val spark = SparkSession.builder()
.appName("Josemtech")
.getOrCreate()
val longAcc = spark.sparkContext.longAccumulator("AcumuladorSuma")
val rdd = spark.sparkContext.parallelize(1 to 25)
rdd.foreach(x => longAcc.add(x))
println(longAcc.value)
El acumulador de tipo long proporciona los siguientes métodos:
- add
- avg
- copy
- copyAndReset
- count
- id
- isRegistered
- isZero
- merge
- name
Spark y Scala en Databricks: Big Data e ingeniería de datos
Trabajo desde niveles básicos hasta avanzados con RDD y DataFrame.
Acumulador double
Para crear una acumulador de tipo double sin nombre debemos utilizar la siguiente sentencia:
val acumulador = spark.sparkContext.doubleAccumulator
Mientras que para crear un acumulador double con nombre debemos utilizar:
val acumulador = spark.sparkContext.doubleAccumulator("NombreDelAcumulador")
Acumulador de colección
Para crear una acumulador de tipo colección sin nombre debemos utilizar la siguiente sentencia:
val acumulador = spark.sparkContext.collectionAccumulator
Mientras que para crear un acumulador de tipo colección con nombre debemos utilizar:
val acumulador = spark.sparkContext.collectionAccumulator("NombreDelAcumulador")
El acumulador de tipo colección proporciona los siguientes métodos:
- add
- copy
- copyAndReset
- id
- isRegistered
- isZero
- merge
- name
- reset
- toString
Ejemplos prácticos
Contar la cantidad de elementos de un Array
val lenguajes = spark.sparkContext.parallelize(Array("Scala", "Python", "C", "Java"))
val cantidadLenguajes = spark.sparkContext.longAccumulator("cantidadLenguajes")
lenguajes.foreach(x => cantidadLenguajes.add(1))
println(cantidadLenguajes.value)
Sumar los elementos de una secuencia de números
val rdd = spark.sparkContext.parallelize(Seq(12.5, 45.4, 23.9, 16))
val suma = spark.sparkContext.doubleAccumulator("sumaSeq")
rdd.foreach(x => suma.add(x))
println(suma.value)
Si desea continuar aprendiendo a trabajar con los RDDs en Scala-Spark y además dominar los DataFrames te invito a que le des un vistazo a mi curso Spark y Scala en Databricks: Big Data e ingeniería de datos en donde podrás encontrar una gran variedad de ejemplos sencillos y prácticos que te ayudarán a consolidar tus conocimientos. Además, el curso cuenta con una serie de ejercicios que te ayudarán a practicar los conocimientos que vayas adquiriendo.
