Yeah, I have observed this common problem in the design a number of times in the mailing list. Tobias, what you linked to is also an additional problem, that occurs with mapPartitions, but not with foreachPartitions (which is relevant here). But I do get your point. I think there was an attempt made in that respect (setup function) using RDD.mapWith and RDD.flatMapWith, but this is got deprecated in 1.0
On Tue, Aug 5, 2014 at 9:54 AM, jschindler <[email protected]> wrote: > You are correct in that I am trying to publish inside of a foreachRDD loop. > I am currently refactoring and will try publishing inside the > foreachPartition loop. Below is the code showing the way it is currently > written, thanks! > > > object myData { > def main(args: Array[String]) { > > val ssc = new StreamingContext("local[8]", "Data", Seconds(10)) > ssc.checkpoint("checkpoint") > val topicMap = Map("pagehit.data" -> 1) > > val factory = new ConnectionFactory() > factory.setUsername("officialUsername") > factory.setPassword("crypticPassword") > factory.setVirtualHost("/") > factory.setHost("rabbit-env") > factory.setPort(0000) > val connection = factory.newConnection() > > val SQLChannel = connection.createChannel() > SQLChannel.queueDeclare("SQLQueue", true, false, false, null) > > val Pipe = KafkaUtils.createStream(ssc, > "Zookeeper_1,Zookeeper_1,Zookeeper_3", "Cons1", > topicMap).map(_._2) > > //PARSE SOME JSON ETC > > windowStream.foreachRDD(pagehit => { > val mongoClient = MongoClient("my-mongodb") > val db = mongoClient("myClient") > val SQLCollection = db("SQLCalls") > > val callArray = pagehit.map(_._1).collect > val avg = (callArray.reduceLeft[Long](_+_))/callArray.length > val URL = pagehit.take(1).map(_._2) > > SQLCollection += MongoDBObject("URL" -> URL(0).substring(7, > URL(0).length - 1), > "Avg Page > Load Time" -> avg) > > val toBuildJSON = Seq(baseMsg, avg.toString, closingBrace) > val byteArray = toBuildJSON.mkString.getBytes() > > SQLChannel.basicPublish("", "SQLQueue", null, byteArray) > > }) > > > > -- > View this message in context: > http://apache-spark-user-list.1001560.n3.nabble.com/Writing-to-RabbitMQ-tp11283p11445.html > Sent from the Apache Spark User List mailing list archive at Nabble.com. > > --------------------------------------------------------------------- > To unsubscribe, e-mail: [email protected] > For additional commands, e-mail: [email protected] > >
