Using spark count on large RDD produces memory error

Viewed 258

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:

  1. 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 .rdd call, solved the memory problem. However, I've ended up with another a bit weird error since I had too many unions.
  2. Investigating that, I've found out that I could load a bunch of files all at once with regex in load method, and thus avoid doing union over all these Datasets. So from ~10 lines of code, I ended up with one which was something like load(pathRegex).map(...).distinct.count().
0 Answers
Related