如何在并发Spark作业中共享RDD

Question

有两种方法可以在两个并发的Spark作业中使用相同的RDD吗？

例如，在下面的应用程序中，我试图将b写入磁盘（作业1），同时计算f（作业2）。但是，Spark似乎一次只能执行一项工作。

val conf = new SparkConf() 
val sc = new SparkContext(conf) 

val a = sc.parallelize(0 until 1000) 
val b = a.mapPartitions(it => { Thread.sleep(5000); it }) 

// Compute b 
b.persist().foreachPartition(_ => {}) 

val c = b.mapPartitions(it => { Thread.sleep(5000); it }) 
val d = c.mapPartitions(it => { Thread.sleep(5000); it }) 
val e = d.mapPartitions(it => { Thread.sleep(5000); it }) 
val f = e.mapPartitions(it => { Thread.sleep(5000); it }) 

// Concurrent actions on b and f (f uses b) 
val actionFuts = List(
    // Job 1 
    Future { 
     Thread.sleep(1000) 
     b.saveAsTextFile("output.ignore/test/b.txt") 
    }, 
    // Job 2 
    Future { 
     f.saveAsTextFile("output.ignore/test/f.txt") 
    } 
) 

Await.result(Future.sequence(actionFuts).map(_ =>()), Duration.Inf) 

Answer 1

已解决。我正在启动4个工作人员的spark-submit，而rdd b有4个分区。因此，所有工作人员都很忙，Spark一次只能执行一项工作（每个工作人员有四个分区）。

为了同时执行这两个作业，我必须使用更多的工人或更少的分区。