火花流：经过ALS

Question

我是新来的火花流。当我在ALS上训练Spark流媒体：它是佩戴的。

java.lang.illegalgumentException：要求失败：在streaming.scala in andursplit上，Mappartitionsrdd [4]中没有可用的评分。Scala：15 \

import org.apache.spark.mllib.recommendation.ALS\

import org.apache.spark.mllib.recommendation.Rating\
import org.apache.spark.SparkConf\
import org.apache.spark.SparkContext\
import org.apache.spark.streaming.{Seconds, StreamingContext}\
import org.apache.spark.streaming._\
object streaming {\
  def main(args: Array[String]) {\
    val conf = new SparkConf().setAppName("ALS").setMaster("local[2]")\
    val ssc = new StreamingContext(conf, Seconds(1))\
    val ratingStream = ssc.textFileStream(directory="/home/chiao/Downloads/streaming/").map(_.split(',') match {case Array(user,item,rate)=>Rating(user.toInt,item.toInt,rate.toInt)})\
    val rank = 100\
    val numIterations = 12\
    val lambda = 0.01\
     ratingStream.foreachRDD(ratingRDD => {val testTrain =  ratingRDD.randomSplit(Array(0.3, 0.7))\
                                           val model = ALS.train(testTrain(1),     rank,numIterations, lambda)\
                                           val test = testTrain(0).map {case Rating(subject, activity, freq) =>(subject, activity)}\
                                           val prediction = model.predict(test)

    })

    ssc.start()
    ssc.awaitTermination
}}

Answer 1

我的数据是：
1,10,100
1,12,100
1,13,100
2,10,100
2,11,100
2,13,100
3,10,100
3,12,100
在文件中另存为user.txt：/home/chiao/downloads/streaming