{"action":"created","issue":{"url":"https://api.github.com/repos/brkyvz/streaming-matrix-factorization/issues/1","repository_url":"https://api.github.com/repos/brkyvz/streaming-matrix-factorization","labels_url":"https://api.github.com/repos/brkyvz/streaming-matrix-factorization/issues/1/labels{/name}","comments_url":"https://api.github.com/repos/brkyvz/streaming-matrix-factorization/issues/1/comments","events_url":"https://api.github.com/repos/brkyvz/streaming-matrix-factorization/issues/1/events","html_url":"https://github.com/brkyvz/streaming-matrix-factorization/issues/1","id":123958381,"number":1,"title":"Exception when starting a stream when no data yet exists (ie. empty Kafka queue)","user":{"login":"cfregly","id":1438064,"avatar_url":"https://avatars.githubusercontent.com/u/1438064?v=3","gravatar_id":"","url":"https://api.github.com/users/cfregly","html_url":"https://github.com/cfregly","followers_url":"https://api.github.com/users/cfregly/followers","following_url":"https://api.github.com/users/cfregly/following{/other_user}","gists_url":"https://api.github.com/users/cfregly/gists{/gist_id}","starred_url":"https://api.github.com/users/cfregly/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cfregly/subscriptions","organizations_url":"https://api.github.com/users/cfregly/orgs","repos_url":"https://api.github.com/users/cfregly/repos","events_url":"https://api.github.com/users/cfregly/events{/privacy}","received_events_url":"https://api.github.com/users/cfregly/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"milestone":null,"comments":2,"created_at":"2015-12-27T03:41:59Z","updated_at":"2016-05-24T04:48:53Z","closed_at":"2016-05-24T04:48:53Z","body":"@brkyvz \r\n\r\nHey burak!\r\n\r\nI'm having an issue where, upon starting my stream when there is no data yet (ie empty Kafka queue), i'm seeing the following error:\r\n\r\n```\r\njava.lang.UnsupportedOperationException: empty collection\r\n\tat org.apache.spark.rdd.RDD$$anonfun$reduce$1$$anonfun$apply$36.apply(RDD.scala:1005)\r\n\tat org.apache.spark.rdd.RDD$$anonfun$reduce$1$$anonfun$apply$36.apply(RDD.scala:1005)\r\n\tat scala.Option.getOrElse(Option.scala:120)\r\n\tat org.apache.spark.rdd.RDD$$anonfun$reduce$1.apply(RDD.scala:1005)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\r\n\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\r\n\tat org.apache.spark.rdd.RDD.reduce(RDD.scala:985)\r\n\tat com.brkyvz.spark.recommendation.LatentMatrixFactorizationModel$.initialize(LatentMatrixFactorizationModel.scala:116)\r\n\tat com.brkyvz.spark.recommendation.StreamingLatentMatrixFactorization$$anonfun$trainOn$1.apply(LatentMatrixFactorization.scala:104)\r\n\tat com.brkyvz.spark.recommendation.StreamingLatentMatrixFactorization$$anonfun$trainOn$1.apply(LatentMatrixFactorization.scala:102)\r\n```\r\nLooks like we might need an empty RDD check here:\r\n\r\nhttps://github.com/brkyvz/streaming-matrix-factorization/blob/master/src/main/scala/com/brkyvz/spark/recommendation/LatentMatrixFactorizationModel.scala#L116\r\n\r\nwhat do you think?\r\n\r\nperhaps you probably tested this with a file-based stream and missed the edge case where no data yet exists upon the first initialize() vall?  Or perhaps I'm not using the API correctly?\r\n\r\nlet me know, man.  excited to get this working.  lots of people asking about this\r\n\r\ni'm currently trying to inject an initial dummy rating as a workaround.\r\n\r\nhere's a code snippet:\r\n\r\n```\r\n    val ratingsStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)\r\n\r\n    val streamingMFParams = new LatentMatrixFactorizationParams()\r\n      .setRank(rank)\r\n      .setIter(maxIterations)\r\n      .setLambda(lambdaRegularization)\r\n\r\n    val streamingMF = new StreamingLatentMatrixFactorization(streamingMFParams)\r\n\r\n    val alsRatingsStream = ratingsStreams.map(message => {\r\n      val tokens = message._2.split(\",\")\r\n      val userId = tokens(0).trim.toLong\r\n      val itemId = tokens(1).trim.toLong\r\n      val rating = tokens(2).trim.toLong\r\n\r\n      Rating(userId, itemId, rating)\r\n    })\r\n\r\n    val streamingMFModel = streamingMF.trainOn(alsRatingsStream)\r\n```"},"comment":{"url":"https://api.github.com/repos/brkyvz/streaming-matrix-factorization/issues/comments/221165772","html_url":"https://github.com/brkyvz/streaming-matrix-factorization/issues/1#issuecomment-221165772","issue_url":"https://api.github.com/repos/brkyvz/streaming-matrix-factorization/issues/1","id":221165772,"user":{"login":"cfregly","id":1438064,"avatar_url":"https://avatars.githubusercontent.com/u/1438064?v=3","gravatar_id":"","url":"https://api.github.com/users/cfregly","html_url":"https://github.com/cfregly","followers_url":"https://api.github.com/users/cfregly/followers","following_url":"https://api.github.com/users/cfregly/following{/other_user}","gists_url":"https://api.github.com/users/cfregly/gists{/gist_id}","starred_url":"https://api.github.com/users/cfregly/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cfregly/subscriptions","organizations_url":"https://api.github.com/users/cfregly/orgs","repos_url":"https://api.github.com/users/cfregly/repos","events_url":"https://api.github.com/users/cfregly/events{/privacy}","received_events_url":"https://api.github.com/users/cfregly/received_events","type":"User","site_admin":false},"created_at":"2016-05-24T04:48:52Z","updated_at":"2016-05-24T04:48:52Z","body":"got this working over in my repo in a more-production-like setting.\r\n\r\nhere are the relevant links:\r\n\r\nNewly-refactored code from this repo:\r\n \r\nhttps://github.com/fluxcapacitor/pipeline/tree/master/myapps/spark/streaming/src/main/scala/com/advancedspark/streaming/ml/incremental\r\n\r\nUse of the code above in a Spark Streaming job pull from Kafka:\r\n\r\nhttps://github.com/fluxcapacitor/pipeline/blob/master/myapps/spark/streaming/src/main/scala/com/advancedspark/streaming/rating/ml/TrainMFIncremental.scala\r\n\r\nclosing this.  thanks so much, burak!"}}
{"actor":{"display_login":"cfregly"}}