Расширенное программирование Spark

Spark содержит два разных типа общих переменных: один - broadcast variables и второй accumulators.

• Broadcast variables - используется для эффективного распределения больших ценностей.

• Accumulators - используется для агрегирования информации конкретной коллекции.

Широковещательные переменные

Широковещательные переменные позволяют программисту хранить переменную, доступную только для чтения, в кэше на каждой машине, а не отправлять ее копию вместе с задачами. Их можно использовать, например, для эффективного предоставления каждому узлу копии большого входного набора данных. Spark также пытается распределять широковещательные переменные, используя эффективные широковещательные алгоритмы, чтобы снизить стоимость связи.

Действия Spark выполняются через набор этапов, разделенных распределенными операциями «перемешивания». Spark автоматически передает общие данные, необходимые для задач на каждом этапе.

Переданные таким образом данные кэшируются в сериализованной форме и десериализуются перед запуском каждой задачи. Это означает, что явное создание широковещательных переменных полезно только тогда, когда для задач на нескольких этапах требуются одни и те же данные или когда важно кэширование данных в десериализованной форме.

Широковещательные переменные создаются из переменной v позвонив SparkContext.broadcast(v). Трансляционная переменная - это оболочка вокругv, и его значение можно получить, вызвав valueметод. Приведенный ниже код показывает это -

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

Output -

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

После создания широковещательной переменной ее следует использовать вместо значения v в любых функциях, выполняемых в кластере, так что vне доставляется в узлы более одного раза. Кроме того, объектv не следует изменять после его широковещательной передачи, чтобы гарантировать, что все узлы получат одинаковое значение широковещательной переменной.

Аккумуляторы

Аккумуляторы - это переменные, которые «добавляются» только посредством ассоциативной операции и поэтому могут эффективно поддерживаться параллельно. Их можно использовать для реализации счетчиков (как в MapReduce) или сумм. Spark изначально поддерживает аккумуляторы числовых типов, и программисты могут добавить поддержку новых типов. Если аккумуляторы созданы с именем, они будут отображаться вSpark’s UI. Это может быть полезно для понимания хода выполнения этапов (ПРИМЕЧАНИЕ - это еще не поддерживается в Python).

Аккумулятор создается из начального значения v позвонив SparkContext.accumulator(v). Задачи, выполняемые в кластере, можно затем добавить к нему с помощьюaddметод или оператор + = (в Scala и Python). Однако они не могут прочитать его значение. Только программа-драйвер может прочитать значение аккумулятора, используя егоvalue метод.

В приведенном ниже коде показан аккумулятор, используемый для суммирования элементов массива:

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

Если вы хотите увидеть вывод вышеуказанного кода, используйте следующую команду -

scala> accum.value

Выход

res2: Int = 10

Числовые операции RDD

Spark позволяет выполнять различные операции с числовыми данными, используя один из предопределенных методов API. Числовые операции Spark реализованы с помощью алгоритма потоковой передачи, который позволяет строить модель по одному элементу за раз.

Эти операции вычисляются и возвращаются как StatusCounter объект, позвонив status() метод.

Ниже приводится список числовых методов, доступных в StatusCounter.

S.No	Методы и значение
1	count() Количество элементов в СДР.
2	Mean() Среднее значение элементов в СДР.
3	Sum() Общая стоимость элементов в СДР.
4	Max() Максимальное значение среди всех элементов в СДР.
5	Min() Минимальное значение среди всех элементов в СДР.
6	Variance() Дисперсия элементов.
7	Stdev() Стандартное отклонение.

Если вы хотите использовать только один из этих методов, вы можете вызвать соответствующий метод непосредственно в RDD.