Исследуя аккумуляторы Spark, я попытался понять и продемонстрировать разницу между аккумулятором и обычной переменной в Spark. Но вывод не соответствует моим ожиданиям. Я имею в виду, что и аккумулятор, и счетчик имеют одинаковое значение в конце программы и могут читать аккумулятор в функции преобразования (согласно документу, только драйвер может читать аккумулятор). Я делаю что-то неправильно? Правильно ли мое понимание?
object Accmulators extends App {
val spark = SparkSession.builder().appName("Accmulators").master("local[3]").getOrCreate()
val cntAccum = spark.sparkContext.longAccumulator("counterAccum")
val file = spark.sparkContext.textFile("resources/InputWithBlank")
var counter = 0
def countBlank(line:String):Array[String]={
val trimmed = line.trim
if(trimmed == "") {
cntAccum.add(1)
cntAccum.value //reading accumulator
counter += 1
}
return line.split(" ")
}
file.flatMap(line => countBlank(line)).collect()
println(cntAccum.value)
println(counter)
}
Входной файл содержит текст с 9 пустыми строками между
4) Aggregations and Joins
5) Spark SQL
6) Spark App tuning
Выход:
И counter, и cntAccum дают одинаковый результат.