Spark AccumulatorParam Общие параметры

Question

У меня проблема с использованием аккумуляторов в Spark. Как видно на веб-сайте Spark, если вы хотите настраивать аккумуляторы, вы можете просто расширить (с помощью объекта) значение AccumulatorParam. Проблема заключается в том, что я хочу, но не могу, сделать этот объект универсальным, например, как это:

object SeqAccumulatorParam[B] extends AccumulatorParam[Seq[B]] { 

    override def zero(initialValue: Seq[B]): Seq[B] = Seq[B]() 

    override def addInPlace(s1: Seq[B], s2: Seq[B]): Seq[B] = s1 ++ s2 

} 

Но это дает мне ошибку компиляции, потому что объекты не могут использовать общие параметры. Моя ситуация не действительно Позвольте мне определить SeqAccumulatorParam для каждого заданного типа, так как это приведет к большому количеству уродливого дублирования кода.

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

Мой вопрос: есть ли другой способ создания аккумуляторов?

Answer 1

Вы можете просто использовать класс для создания объектов вместо одноэлементных объектов.

class SeqAccumulatorParam[B] extends AccumulatorParam[Seq[B]] { 
    override def zero(initialValue: Seq[B]): Seq[B] = Seq[B]() 
    override def addInPlace(s1: Seq[B], s2: Seq[B]): Seq[B] = s1 ++ s2 
} 

val seqAccum = sc.accumulator(Seq[Int]())(new SeqAccumulatorParam[Int]()) 

val lists = (1 to 5).map(x => (0 to x).toList) 
sc.parallelize(lists).foreach(x => seqAccum += x) 

seqAccum.value 
// Seq[Int] = List(0, 1, 2, 3, 4, 0, 1, 2, 3, 4, 5, 0, 1, 2, 3, 0, 1, 2, 0, 1) 
// result can be in different order. 

// For Doubles. 
val seqAccumD = sc.accumulator(Seq[Double]())(new SeqAccumulatorParam[Double]()) 
sc.parallelize(lists.map(x => x.map(_.toDouble))).foreach(x => seqAccumD += x) 

seqAccumD.value 
// Seq[Double] = List(0.0, 1.0, 0.0, 1.0, 2.0, 0.0, 1.0, 2.0, 3.0, 0.0, 1.0, 2.0, 3.0, 4.0, 0.0, 1.0, 2.0, 3.0, 4.0, 5.0)