Programación avanzada de Spark

Spark contiene dos tipos diferentes de variables compartidas: una es broadcast variables y el segundo es accumulators.

• Broadcast variables - utilizado para distribuir de manera eficiente valores grandes.

• Accumulators - utilizado para agregar la información de una colección particular.

Variables de transmisión

Las variables de difusión permiten al programador mantener una variable de solo lectura almacenada en caché en cada máquina en lugar de enviar una copia de ella con tareas. Se pueden usar, por ejemplo, para dar a cada nodo, una copia de un gran conjunto de datos de entrada, de una manera eficiente. Spark también intenta distribuir variables de transmisión utilizando algoritmos de transmisión eficientes para reducir el costo de comunicación.

Las acciones de Spark se ejecutan a través de un conjunto de etapas, separadas por operaciones distribuidas "shuffle". Spark transmite automáticamente los datos comunes que necesitan las tareas dentro de cada etapa.

Los datos difundidos de esta manera se almacenan en caché en forma serializada y se deserializan antes de ejecutar cada tarea. Esto significa que la creación explícita de variables de difusión solo es útil cuando las tareas en varias etapas necesitan los mismos datos o cuando es importante almacenar en caché los datos en forma deserializada.

Las variables de difusión se crean a partir de una variable v llamando SparkContext.broadcast(v). La variable de difusión es un envoltoriov, y se puede acceder a su valor llamando al valuemétodo. El código que se proporciona a continuación muestra esto:

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))

Output -

broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)

Una vez creada la variable de difusión, debe utilizarse en lugar del valor v en cualquier función que se ejecute en el clúster, de modo que vno se envía a los nodos más de una vez. Además, el objetov no debe modificarse después de su difusión, para garantizar que todos los nodos obtengan el mismo valor de la variable de difusión.

Acumuladores

Los acumuladores son variables que solo se “agregan” a través de una operación asociativa y, por lo tanto, pueden ser soportadas eficientemente en paralelo. 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 soporte para nuevos tipos. Si los acumuladores se crean con un nombre, se mostrarán enSpark’s UI. Esto puede ser útil para comprender el progreso de las etapas de ejecución (NOTA: esto aún no es compatible con Python).

Se crea un acumulador a partir de un valor inicial v llamando SparkContext.accumulator(v). Las tareas que se ejecutan en el clúster se pueden agregar mediante eladdo el operador + = (en Scala y Python). Sin embargo, no pueden leer su valor. Solo el programa del controlador puede leer el valor del acumulador, utilizando suvalue método.

El código que se proporciona a continuación muestra un acumulador que se usa para sumar los elementos de una matriz:

scala> val accum = sc.accumulator(0) 
 
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)

Si desea ver el resultado del código anterior, use el siguiente comando:

scala> accum.value

Salida

res2: Int = 10

Operaciones RDD numéricas

Spark le permite realizar diferentes operaciones con datos numéricos, utilizando uno de los métodos API predefinidos. Las operaciones numéricas de Spark se implementan con un algoritmo de transmisión que permite construir el modelo, un elemento a la vez.

Estas operaciones se calculan y devuelven como StatusCounter objeto llamando status() método.

La siguiente es una lista de métodos numéricos disponibles en StatusCounter.

S. No	Métodos y significado
1	count() Número de elementos en el RDD.
2	Mean() Promedio de los elementos del RDD.
3	Sum() Valor total de los elementos del RDD.
4	Max() Valor máximo entre todos los elementos del RDD.
5	Min() Valor mínimo entre todos los elementos del RDD.
6	Variance() Varianza de los elementos.
7	Stdev() Desviación Estándar.

Si desea utilizar solo uno de estos métodos, puede llamar al método correspondiente directamente en RDD.