Apache Spark中的shuffle read和shuffle write是什么？

Question

在端口8080上运行的Spark管理员的屏幕截图如下所示：

在此代码中，“Shuffle读取”和“Shuffle写入”参数始终为空：

import org.apache.spark.SparkContext;

object first {
  println("Welcome to the Scala worksheet")

  val conf = new org.apache.spark.SparkConf()
    .setMaster("local")
    .setAppName("distances")
    .setSparkHome("C:\\spark-1.1.0-bin-hadoop2.4\\spark-1.1.0-bin-hadoop2.4")
    .set("spark.executor.memory", "2g")
  val sc = new SparkContext(conf)

  def euclDistance(userA: User, userB: User) = {

    val subElements = (userA.features zip userB.features) map {
      m => (m._1 - m._2) * (m._1 - m._2)
    }
    val summed = subElements.sum
    val sqRoot = Math.sqrt(summed)

    println("value is" + sqRoot)
    ((userA.name, userB.name), sqRoot)
  }

  case class User(name: String, features: Vector[Double])

  def createUser(data: String) = {

    val id = data.split(",")(0)
    val splitLine = data.split(",")

    val distanceVector = (splitLine.toList match {
      case h :: t => t
    }).map(m => m.toDouble).toVector

    User(id, distanceVector)

  }

  val dataFile = sc.textFile("c:\\data\\example.txt")
  val users = dataFile.map(m => createUser(m))
  val cart = users.cartesian(users) //
  val distances = cart.map(m => euclDistance(m._1, m._2))
  //> distances  : org.apache.spark.rdd.RDD[((String, String), Double)] = MappedR
  //| DD[4] at map at first.scala:46
  val d = distances.collect //

  d.foreach(println) //> ((a,a),0.0)
  //| ((a,b),0.0)
  //| ((a,c),1.0)
  //| ((a,),0.0)
  //| ((b,a),0.0)
  //| ((b,b),0.0)
  //| ((b,c),1.0)
  //| ((b,),0.0)
  //| ((c,a),1.0)
  //| ((c,b),1.0)
  //| ((c,c),0.0)
  //| ((c,),0.0)
  //| ((,a),0.0)
  //| ((,b),0.0)
  //| ((,c),0.0)
  //| ((,),0.0)

}

为什么“Shuffle Read”和“Shuffle Write”字段为空？上述代码能否进行调整以填充这些字段，以便了解如何实现？

- blue-sky

Answer 1

洗牌指的是在多个Spark阶段之间重新分配数据。 "Shuffle Write" 意思是在传输之前（通常在一个阶段结束时），所有执行程序上编写的序列化数据的总和，而 "Shuffle Read" 意思是在一个阶段开始时，在所有执行程序上读取序列化数据的总和。

您的程序只有一个阶段，由 "collect" 操作触发。不需要进行洗牌，因为您只有一堆连续的映射操作，这些操作在一个阶段中被管道化。

请尝试查看这些幻灯片：http://de.slideshare.net/colorant/spark-shuffle-introduction

阅读原始论文的第5章也可能会有帮助：http://people.csail.mit.edu/matei/papers/2012/nsdi_spark.pdf

- taruxtin

Answer 2

我认为您需要以集群/分布式模式运行应用程序才能查看任何Shuffle读取或写入值。通常情况下，“shuffle”是由Spark操作的子集触发的（例如，groupBy、join等）。

- Soumya Simanta