Stackoverflow due to long RDD Lineage

by

In general you can use checkpoints to break long lineages. Some more or less similar to this should work:

import org.apache.spark.rdd.RDD
import scala.reflect.ClassTag

val checkpointInterval: Int = ???

def loadAndFilter(path: String) = sc.textFile(path)
  .filter(_.startsWith("#####"))
  .map((path, _))

def mergeWithLocalCheckpoint[T: ClassTag](interval: Int)
  (acc: RDD[T], xi: (RDD[T], Int)) = {
    if(xi._2 % interval == 0 & xi._2 > 0) xi._1.union(acc).localCheckpoint
    else xi._1.union(acc)
  }

val zero: RDD[(String, String)] = sc.emptyRDD[(String, String)]
fileList.map(loadAndFilter).zipWithIndex
  .foldLeft(zero)(mergeWithLocalCheckpoint(checkpointInterval))

In this particular situation a much simpler solution should be to use SparkContext.union method:

val masterRDD = sc.union(
  fileList.map(path => sc.textFile(path)
    .filter(_.startsWith("#####"))
    .map((path, _))) 
)

A difference between these methods should be obvious when you take a look at the DAG generated by loop / reduce:

enter image description here

and a single union:

enter image description here

Of course if files are small you can combine wholeTextFiles with flatMap and read all files at once:

sc.wholeTextFiles(fileList.mkString(","))
  .flatMap{case (path, text) =>  
    text.split("\n").filter(_.startsWith("#####")).map((path, _))}

More Related Contents:

  1. How does HashPartitioner work?
  2. How to convert rdd object to dataframe in spark
  3. (Why) do we need to call cache or persist on a RDD
  4. Case class equality in Apache Spark
  5. Spark performance for Scala vs Python
  6. How to find spark RDD/Dataframe size?
  7. Modify collection inside a Spark RDD foreach
  8. Parsing multiline records in Scala
  9. Explanation of fold method of spark RDD
  10. Why does partition parameter of SparkContext.textFile not take effect?
  11. Why does Spark RDD partition has 2GB limit for HDFS?
  12. Reading in multiple files compressed in tar.gz archive into Spark [duplicate]
  13. How to sort an RDD in Scala Spark?
  14. How to transpose an RDD in Spark
  15. How to define and use a User-Defined Aggregate Function in Spark SQL?
  16. How to load local file in sc.textFile, instead of HDFS
  17. Spark read file from S3 using sc.textFile (“s3n://…)
  18. Stratified sampling in Spark
  19. How do I convert csv file to rdd
  20. Caused by: java.lang.NullPointerException at org.apache.spark.sql.Dataset
  21. How to save DataFrame directly to Hive?
  22. About how to add a new column to an existing DataFrame with random values in Scala
  23. How to write unit tests in Spark 2.0+?
  24. Customize SparkContext using sparkConf.set(..) when using spark-shell
  25. Spark: Transpose DataFrame Without Aggregating
  26. Exploding nested Struct in Spark dataframe
  27. How do I iterate RDD’s in apache spark (scala)
  28. How to compute cumulative sum using Spark
  29. Add Number of days column to Date Column in same dataframe for Spark Scala App
  30. Spark – Error “A master URL must be set in your configuration” when submitting an app

Leave a Comment