Stackoverflow из-за длительной RDD Lineage

Ответ 1

В общем, вы можете использовать контрольные точки для разрыва длинных линий. Некоторые более или менее похожие на это должны работать:

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))

В этой конкретной ситуации гораздо более простым решением должно быть использование метода SparkContext.union:

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

Разница между этими методами должна быть очевидной, когда вы смотрите на DAG, сгенерированную циклом / reduce:

и один union:

Конечно, если файлы маленькие, вы можете объединить wholeTextFiles с flatMap и сразу прочитать все файлы:

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