Spark accumulableCollection does not work with mutable.Map
I use Spark to collect employee savings and for that I use a Spark battery. I am using Map [empId, emp] as an accumulableCollection so that I can search for an employee by their IDs. I tried everything but it doesn't work. Can anyone point out if there are any logical problems with the way I use accumulableCollection or Map is not supported. Below is my code
package demo
import org.apache.spark.{SparkContext, SparkConf, Logging}
import org.apache.spark.SparkContext._
import scala.collection.mutable
object MapAccuApp extends App with Logging {
case class Employee(id:String, name:String, dept:String)
val conf = new SparkConf().setAppName("Employees") setMaster ("local[4]")
val sc = new SparkContext(conf)
implicit def empMapToSet(empIdToEmp: mutable.Map[String, Employee]): mutable.MutableList[Employee] = {
empIdToEmp.foldLeft(mutable.MutableList[Employee]()) { (l, e) => l += e._2}
}
val empAccu = sc.accumulableCollection[mutable.Map[String, Employee], Employee](mutable.Map[String,Employee]())
val employees = List(
Employee("10001", "Tom", "Eng"),
Employee("10002", "Roger", "Sales"),
Employee("10003", "Rafael", "Sales"),
Employee("10004", "David", "Sales"),
Employee("10005", "Moore", "Sales"),
Employee("10006", "Dawn", "Sales"),
Employee("10007", "Stud", "Marketing"),
Employee("10008", "Brown", "QA")
)
System.out.println("employee count " + employees.size)
sc.parallelize(employees).foreach(e => {
empAccu += e
})
System.out.println("empAccumulator size " + empAccu.value.size)
}
source to share
Usage accumulableCollection
seems overkill for your problem like below:
import org.apache.spark.{AccumulableParam, Accumulable, SparkContext, SparkConf}
import scala.collection.mutable
case class Employee(id:String, name:String, dept:String)
val conf = new SparkConf().setAppName("Employees") setMaster ("local[4]")
val sc = new SparkContext(conf)
implicit def mapAccum =
new AccumulableParam[mutable.Map[String,Employee], Employee]
{
def addInPlace(t1: mutable.Map[String,Employee],
t2: mutable.Map[String,Employee])
: mutable.Map[String,Employee] = {
t1 ++= t2
t1
}
def addAccumulator(t1: mutable.Map[String,Employee], e: Employee)
: mutable.Map[String,Employee] = {
t1 += (e.id -> e)
t1
}
def zero(t: mutable.Map[String,Employee])
: mutable.Map[String,Employee] = {
mutable.Map[String,Employee]()
}
}
val empAccu = sc.accumulable(mutable.Map[String,Employee]())
val employees = List(
Employee("10001", "Tom", "Eng"),
Employee("10002", "Roger", "Sales"),
Employee("10003", "Rafael", "Sales"),
Employee("10004", "David", "Sales"),
Employee("10005", "Moore", "Sales"),
Employee("10006", "Dawn", "Sales"),
Employee("10007", "Stud", "Marketing"),
Employee("10008", "Brown", "QA")
)
System.out.println("employee count " + employees.size)
sc.parallelize(employees).foreach(e => {
empAccu += e
})
println("empAccumulator size " + empAccu.value.size)
empAccu.value.foreach(entry =>
println("emp id = " + entry._1 + " name = " + entry._2.name))
While this is poorly documented, the relevant test in the Spark codebase is fairly covered.
Edit: It turns out that usage accumulableCollection
does matter: you don't need to define AccumulableParam
, and the following works. I leave both solutions in case they are helpful to people.
case class Employee(id:String, name:String, dept:String)
val conf = new SparkConf().setAppName("Employees") setMaster ("local[4]")
val sc = new SparkContext(conf)
val empAccu = sc.accumulableCollection(mutable.HashMap[String,Employee]())
val employees = List(
Employee("10001", "Tom", "Eng"),
Employee("10002", "Roger", "Sales"),
Employee("10003", "Rafael", "Sales"),
Employee("10004", "David", "Sales"),
Employee("10005", "Moore", "Sales"),
Employee("10006", "Dawn", "Sales"),
Employee("10007", "Stud", "Marketing"),
Employee("10008", "Brown", "QA")
)
System.out.println("employee count " + employees.size)
sc.parallelize(employees).foreach(e => {
// notice this is different from the previous solution
empAccu += e.id -> e
})
println("empAccumulator size " + empAccu.value.size)
empAccu.value.foreach(entry =>
println("emp id = " + entry._1 + " name = " + entry._2.name))
Both solutions were tested using Spark 1.0.2.
source to share