我有一个RDD,格式如下:byUserHour: org.apache.spark.rdd.RDD[(String, String, Int)]
我想要创建这些数据的稀疏矩阵,用于计算中位数、平均值等。RDD包含行ID、列ID和值。我有两个数组,其中包含用于查找的行ID和列ID字符串。
这是我的尝试:
import breeze.linalg._
val builder = new CSCMatrix.Builder[Int](rows=BCnUsers.value.toInt,cols=broadcastTimes.value.size)
byUserHour.foreach{x =>
val row = userids.indexOf(x._1)
val col = broadcastTimes.value.indexOf(x._2)
builder.add(row,col,x._3)}
builder.result()
这是我的错误信息:
14/06/10 16:39:34 INFO DAGScheduler: Failed to run foreach at <console>:38
org.apache.spark.SparkException: Job aborted due to stage failure: Task not serializable: java.io.NotSerializableException: breeze.linalg.CSCMatrix$Builder$mcI$sp
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1033)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1017)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1015)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1015)
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$submitMissingTasks(DAGScheduler.scala:770)
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$submitStage(DAGScheduler.scala:713)
at org.apache.spark.scheduler.DAGScheduler.handleJobSubmitted(DAGScheduler.scala:697)
at org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1176)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
at akka.actor.ActorCell.invoke(ActorCell.scala:456)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
at akka.dispatch.Mailbox.run(Mailbox.scala:219)
at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
我的数据集非常大,所以如果可能的话,我希望进行分布式处理。如果有任何帮助,将不胜感激。
进展更新:
CSCMartix不能在Spark中工作。然而,有一个扩展了DistributedMatrix
的RowMatrix。 RowMatrix
确实有一个方法computeColumnSummaryStatistics()
,应该能够计算出我正在寻找的一些统计信息。我知道MLlib每天都在发展,所以我会关注更新,但与此同时,我将尝试创建一个RDD [Vector]
来输入到RowMatrix
中。需要注意的是,RowMatrix
是实验性的,并且表示没有有意义的行索引的面向行的分布式矩阵。
SparseVector
实现:http://people.csail.mit.edu/matei/spark-unified-docs/api/scala/index.html#org.apache.spark.mllib.linalg.SparseVector - maasg