Re: new user question on using scala collections inside RDDs

[Ewen Cheslack-Postava]

Code in a transformation (i.e. inside the function passed to RDD.map() or RDD.filter()) will run on workers, not the driver. They will run in parallel. In Spark, the driver actually doesn't do much -- it just builds up a description of the computation to be performed and then sends it off to the cluster master when you perform an action that actually triggers computation (so long as you use RDD transformations to do most of the work). In your example, the bold items are running on workers and are properly distributed since they all occur in the transformation functions. As for Scala collections in RDDs, my impression is that the problems usually arise with respect to the size of the data. A Seq may take much more memory than a plain old Array. In some cases it is worth it to make sure you store the right type. That said, keep in mind that this is really only an issue if you're *caching* or *shuffling* that data. For example, if you load data and then do a map() and filter() and a map(), those will all be processed together because none of them require a shuffle. Any *intermediate* types won't be stores, just processed and garbage collected. Unless you generate a lot of garbage, these probably won't be a major issue. Personally, I tend to use Scala collections as usual, but then pay attention to bottlenecks at shuffle and cache points in the pipeline. In some cases it makes sense to make sure I use Arrays instead of a Seq. But these are the same type of optimizations you should probably be making at the time you decide whether to store data serialized or unserialized when caching, i.e. you probably shouldn't worry about it until you actually have some profiling data in hand to inform your decisions. -Ewen Peter March 14, 2014 at 7:12 PM Hi  I'm new to Spark. I have played with some data locally but starting to wonder if I'm going down a wrong track of using Scala collections inside RDDs.  I'm looking at a log file of events from mobile clients. One of the engagement metrics we're interested in is lifetime (not terribly interesting on its own but reused in other metrics). To calculate this I've written the following: val events = csvFile.map(line => line.split(",")) val sessionEventsByUsers = events.map(x => (x._1, x._2.filter(e => "session.start".equals(e(0)) || "session.end".equals(e(0)) ))).groupBy(_(2)) this gives me:    RDD[(String, Seq[Array[String]])]   which is:    user id -> Seq[Array(event, timestamp, userid, sessionid)] sessionEventsByUsers.map(x => (x._1, x._2.map(_(1).toLong).sorted)).map(x => (x._1, x._2.last - x._2.head)) this gives us our results   user id -> lifetime  I'm wondering if using Scala collections inside the RDD like this is a good idea, from a brief glance at the tuning guide I'm cautious. In this contrived case a user could potentially have 1000s of events, I'm wondering what happens behind the scenes, does the bolded operations on the nested scala collections get distributed to worker nodes or does it all run on the driver node (ie, not horizontally scalable). I'm hopeful for the former and suspect that's the case (I found that seq.grouped(x) doesn't work to scala.collector.Iterator not being serializable).  Is there a way to do something like this without using the nested scala collections & transformations? Don't know any tricks yet :) Thank you! Peter