最简单程序的大任务规模

Question

我试图用Spark运行最简单的程序

import org.apache.spark.{SparkContext, SparkConf}

object LargeTaskTest {

  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("DataTest").setMaster("local[*]")
    val sc = new SparkContext(conf)
    val dat = (1 to 10000000).toList
    val data = sc.parallelize(dat).cache()
    for(i <- 1 to 100){
      println(data.reduce(_ + _))
    }
  }   
}

每次迭代后,我收到以下错误消息:

WARN TaskSetManager:阶段0包含一个非常大的任务(9767 KB).建议的最大任务大小为100 KB.

增加数据大小会增加所述任务大小.这告诉我,驱动程序正在向所有执行程序发送"dat"对象,但我不能为我的生活看到原因,因为我的RDD上的唯一操作是reduce,它基本上没有关闭.有任何想法吗 ？

Answer 1

因为您首先在本地创建非常大的列表,所以Spark parallelize方法尝试将此列表作为单个单元发送给Spark工作器,作为任务的一部分.因此,您收到警告信息.作为替代方案,您可以并行化一个小得多的列表,然后使用flatMap它将其分解为更大的列表.这也有利于并行创建更大的数字集.例如:

import org.apache.spark.{SparkContext, SparkConf}

object LargeTaskTest extends App {

  val conf = new SparkConf().setAppName("DataTest").setMaster("local[*]")
  val sc = new SparkContext(conf)
  val dat = (0 to 99).toList
  val data = sc.parallelize(dat).cache().flatMap(i => (1 to 1000000).map(j => j * 100 + i))
  println(data.count()) //100000000
  println(data.reduce(_ + _))
  sc.stop()
}

编辑:

最终,并行化的本地集合必须被推送给执行者.该parallelize方法创建ParallelCollectionRDD的实例:

def parallelize[T: ClassTag](
      seq: Seq[T],
      numSlices: Int = defaultParallelism): RDD[T] = withScope {
    assertNotStopped()
    new ParallelCollectionRDD[T](this, seq, numSlices, Map[Int, Seq[String]]())
  }

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/SparkContext.scala#L730

ParallelCollectionRDD创建了许多等于的分区numSlices:

  override def getPartitions: Array[Partition] = {
    val slices = ParallelCollectionRDD.slice(data, numSlices).toArray
    slices.indices.map(i => new ParallelCollectionPartition(id, i, slices(i))).toArray
  }

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/ParallelCollectionRDD.scala#L96

numSlicessc.defaultParallelism我的机器上的默认值是4.所以即使拆分,每个分区都包含一个非常大的列表,需要将其推送到执行程序.

SparkContext.parallelize包含注释@note Parallelize acts lazily并ParallelCollectionRDD包含注释;

// TODO:现在,每个拆分都会发送完整的数据,即使稍后在RDD链中也会缓存//缓存.将数据写入DFS中的文件并在分割//中读取它可能是值得的.

所以看起来当你调用reduce时会出现问题,因为这是分区被发送到执行程序的重点,但根本原因是你在一个非常大的列表上调用并行化.在执行器中生成大型列表是一种更好的方法,恕我直言.