I'm doing a union over ~100 RDDs, that cumulatively take several GBs. In the end, I want to do a count, and count how many of these unique Strings do I have across all these files. However, zeppelin breaks with a memory error:
org.apache.spark.SparkException: Job aborted due to stage failure: Total size of serialized results of 8263 tasks (1024.1 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:2041)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2029)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2028)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2028)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:966)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:966)
at scala.Option.foreach(Option.scala:257)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:966)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2262)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2211)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2200)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:777)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2061)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2082)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2101)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2126)
at org.apache.spark.rdd.RDD.count(RDD.scala:1168)
... 55 elided
Now, I'm aware that count is an action, but shouldn't it be optimized not to pull all the data locally, and calculate the count in a distributed fashion? What am I missing?
Note: I'm not using any collect-s, only select, distinct, map and union when it comes to transformations, and count from actions.
UPDATE:
So I spent some time investigating and googling how to solve this. Essentially, I found two improvements that helped me out:
- I've found out that using RDDs instead of Datasets is greatly suboptimal if there's no super specific reason why would someone convert. This change only, which was pretty much just removing the
.rddcall, solved the memory problem. However, I've ended up with another a bit weird error since I had too manyunions. - Investigating that, I've found out that I could load a bunch of files all at once with regex in
loadmethod, and thus avoid doingunionover all these Datasets. So from ~10 lines of code, I ended up with one which was something likeload(pathRegex).map(...).distinct.count().