spark正在单个执行器上运行具有多个分区的作业

Question

我有一个Spark 2.1.1作业，正在Mesos集群中运行。Spark UI显示32个活动执行器和RDD。getNumPartitions显示28个分区。但只有一个（随机）执行者在做任何工作，所有其他执行者都标记为已完成。我向执行器代码（stdout）添加了调试语句，只有一个执行器显示这些语句。整个管道的结构如下：获取ID列表-

stage 1: val ids=session.sparkContext.textFile(path).repartition(28) -> RDD[String]

//ids.getNumPartitions shows 28
stage 2: val json=ids.mapPartitions { keys =>
  val urlBuilder ...
  val buffer ....
  keys map { key =>
    val url=urlBuilder.createUrl(id) //java.net.URL
    val json=url.openStream() ... //download text to buffer, close stream
    (id,json.toString)
  }
} -> RDD[Tuple2[String,String]]

stage 3: val output = json flatMap { t =>
  val values = ... //parse JSON, get values from JSON or empty sequence if not found
  values map { value => (t._1, value) }
} -> RDD[Tuple2[String,String]]

stage 4: output.saveAsTextFile("s3://...")

这些是Spark二进制文件的配置设置：--drive-内存32g--confspark.driver.cores=4--execort-内存4g--confspark.cores.max=128--confspark.executor.cores=4

仅在一个执行器上运行的阶段是第二个阶段。我在第一步中明确指定了分区数（重新分区（28））。以前有人见过这种行为吗？谢谢

M

解决方案

我走了另一条路（参见Travis的建议），将分区的数量（在步骤1之后）增加到100个。这项工作奏效了，几分钟内就完成了。但是有一个副作用——现在我有100个部分文件放在S3中。

Answer 1

确保在“获取id列表”之后发生您的. re分区（）阶段。

听起来像是先生成一个包含28个分区的空集，然后将ID列表放入一个分区中。

提供示例代码后编辑：

是否可能每个任务都很快完成（即在几秒钟内）？我见过spark在短时间内完成任务时，即使有数千个未完成的任务，也不会将任务安排给空闲的执行者。如果是这样的话，您可能需要更少的分区来让每个任务花费更长的时间。有时，这足以触发任务调度器将更多任务调度给空闲的执行者。