Spark Shuffle过程分析:Map阶段处理流程

标签: Spark 开源技术 Spark-2.0.0 | 发表时间:2017-05-26 15:13 | 作者:Yanjun
分享到:
出处:http://shiyanjun.cn

默认配置情况下,Spark在Shuffle过程中会使用SortShuffleManager来管理Shuffle过程中需要的基本组件,以及对RDD各个Partition数据的计算。我们可以在Driver和Executor对应的SparkEnv对象创建过程中看到对应的配置,如下代码所示:

    // Let the user specify short names for shuffle managers
    val shortShuffleMgrNames = Map(
      "sort" -> classOf[org.apache.spark.shuffle.sort.SortShuffleManager].getName,
      "tungsten-sort" -> classOf[org.apache.spark.shuffle.sort.SortShuffleManager].getName)
    val shuffleMgrName = conf.get("spark.shuffle.manager", "sort")
    val shuffleMgrClass = shortShuffleMgrNames.getOrElse(shuffleMgrName.toLowerCase, shuffleMgrName)
    val shuffleManager = instantiateClass[ShuffleManager](shuffleMgrClass)

如果需要修改ShuffleManager实现,则只需要修改配置项spark.shuffle.manager即可,默认支持sort和 tungsten-sort,可以指定自己实现的ShuffleManager类。
因为Shuffle过程中需要将Map结果数据输出到文件,所以需要通过注册一个ShuffleHandle来获取到一个ShuffleWriter对象,通过它来控制Map阶段记录数据输出的行为。其中,ShuffleHandle包含了如下基本信息:

  • shuffleId:标识Shuffle过程的唯一ID
  • numMaps:RDD对应的Partitioner指定的Partition的个数,也就是ShuffleMapTask输出的Partition个数
  • dependency:RDD对应的依赖ShuffleDependency

下面我们看下,在SortShuffleManager中是如何注册Shuffle的,代码如下所示:

  override def registerShuffle[K, V, C](
      shuffleId: Int,
      numMaps: Int,
      dependency: ShuffleDependency[K, V, C]): ShuffleHandle = {
    if (SortShuffleWriter.shouldBypassMergeSort(SparkEnv.get.conf, dependency)) {
      new BypassMergeSortShuffleHandle[K, V](
        shuffleId, numMaps, dependency.asInstanceOf[ShuffleDependency[K, V, V]])
    } else if (SortShuffleManager.canUseSerializedShuffle(dependency)) {
      new SerializedShuffleHandle[K, V](
        shuffleId, numMaps, dependency.asInstanceOf[ShuffleDependency[K, V, V]])
    } else {
      new BaseShuffleHandle(shuffleId, numMaps, dependency)
    }
  }

上面代码中,对应如下3种ShuffleHandle可以选择,说明如下:

  • BypassMergeSortShuffleHandle

如果dependency不需要进行Map Side Combine,并且RDD对应的ShuffleDependency中的Partitioner设置的Partition的数量(这个不要和parent RDD的Partition个数混淆,Partitioner指定了map处理结果的Partition个数,每个Partition数据会在Shuffle过程中全部被拉取而拷贝到下游的某个Executor端)小于等于配置参数spark.shuffle.sort.bypassMergeThreshold的值,则会注册BypassMergeSortShuffleHandle。默认情况下,spark.shuffle.sort.bypassMergeThreshold的取值是200,这种情况下会直接将对RDD的 map处理结果的各个Partition数据写入文件,并最后做一个合并处理。

  • SerializedShuffleHandle

如果ShuffleDependency中的Serializer,允许对将要输出数据对象进行排序后,再执行序列化写入到文件,则会选择创建一个SerializedShuffleHandle。

  • BaseShuffleHandle

除了上面两种ShuffleHandle以后,其他情况都会创建一个BaseShuffleHandle对象,它会以反序列化的格式处理Shuffle输出数据。

Map阶段处理流程分析

Map阶段RDD的计算,对应ShuffleMapTask这个实现类,它最终会在每个Executor上启动运行,每个ShuffleMapTask处理RDD的一个Partition的数据。这个过程的核心处理逻辑,代码如下所示:

      val manager = SparkEnv.get.shuffleManager
      writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId, context)
      writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])

上面代码中,在调用rdd的iterator()方法时,会根据RDD实现类的compute方法指定的处理逻辑对数据进行处理,当然,如果该Partition对应的数据已经处理过并存储在MemoryStore或DiskStore,直接通过BlockManager获取到对应的Block数据,而无需每次需要时重新计算。然后,write()方法会将已经处理过的Partition数据输出到磁盘文件。
在Spark Shuffle过程中,每个ShuffleMapTask会通过配置的ShuffleManager实现类对应的ShuffleManager对象(实际上是在SparkEnv中创建),根据已经注册的ShuffleHandle,获取到对应的ShuffleWriter对象,然后通过ShuffleWriter对象将Partition数据写入内存或文件。所以,接下来我们可能关心每一种ShuffleHandle对应的ShuffleWriter的行为,可以看到SortShuffleManager中获取到ShuffleWriter的实现代码,如下所示:

  /** Get a writer for a given partition. Called on executors by map tasks. */
  override def getWriter[K, V](
      handle: ShuffleHandle,
      mapId: Int,
      context: TaskContext): ShuffleWriter[K, V] = {
    numMapsForShuffle.putIfAbsent(
      handle.shuffleId, handle.asInstanceOf[BaseShuffleHandle[_, _, _]].numMaps)
    val env = SparkEnv.get
    handle match {
      case unsafeShuffleHandle: SerializedShuffleHandle[K @unchecked, V @unchecked] =>
        new UnsafeShuffleWriter(
          env.blockManager,
          shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver],
          context.taskMemoryManager(),
          unsafeShuffleHandle,
          mapId,
          context,
          env.conf)
      case bypassMergeSortHandle: BypassMergeSortShuffleHandle[K @unchecked, V @unchecked] =>
        new BypassMergeSortShuffleWriter(
          env.blockManager,
          shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver],
          bypassMergeSortHandle,
          mapId,
          context,
          env.conf)
      case other: BaseShuffleHandle[K @unchecked, V @unchecked, _] =>
        new SortShuffleWriter(shuffleBlockResolver, other, mapId, context)
    }
  }

我们以最简单的SortShuffleWriter为例进行分析,在SortShuffleManager可以通过getWriter()方法创建一个SortShuffleWriter对象,然后在ShuffleMapTask中调用SortShuffleWriter对象的write()方法处理Map输出的记录数据,write()方法的处理代码,如下所示:

  /** Write a bunch of records to this task's output */
  override def write(records: Iterator[Product2[K, V]]): Unit = {
    sorter = if (dep.mapSideCombine) {
      require(dep.aggregator.isDefined, "Map-side combine without Aggregator specified!")
      new ExternalSorter[K, V, C](
        context, dep.aggregator, Some(dep.partitioner), dep.keyOrdering, dep.serializer)
    } else {
      // In this case we pass neither an aggregator nor an ordering to the sorter, because we don't
      // care whether the keys get sorted in each partition; that will be done on the reduce side
      // if the operation being run is sortByKey.
      new ExternalSorter[K, V, V](
        context, aggregator = None, Some(dep.partitioner), ordering = None, dep.serializer)
    }
    sorter.insertAll(records)

    // Don't bother including the time to open the merged output file in the shuffle write time,
    // because it just opens a single file, so is typically too fast to measure accurately
    // (see SPARK-3570).
    val output = shuffleBlockResolver.getDataFile(dep.shuffleId, mapId)
    val tmp = Utils.tempFileWith(output)
    val blockId = ShuffleBlockId(dep.shuffleId, mapId, IndexShuffleBlockResolver.NOOP_REDUCE_ID)
    val partitionLengths = sorter.writePartitionedFile(blockId, tmp)
    shuffleBlockResolver.writeIndexFileAndCommit(dep.shuffleId, mapId, partitionLengths, tmp)
    mapStatus = MapStatus(blockManager.shuffleServerId, partitionLengths)
  }

从SortShuffleWriter类中的write()方法可以看到,最终调用了ExeternalSorter的insertAll()方法,实现了Map端RDD某个Partition数据处理并输出到内存或磁盘文件,这也是处理Map阶段输出记录数据最核心、最复杂的过程。我们将其分为两个阶段进行分析:第一阶段是,ExeternalSorter的insertAll()方法处理过程,将记录数据Spill到磁盘文件;第二阶段是,执行完insertAll()方法之后的处理逻辑,创建Shuffle Block数据文件及其索引文件。

内存缓冲写记录数据并Spill到磁盘文件

查看SortShuffleWriter类的write()方法可以看到,在内存中缓存记录数据的数据结构有两种:一种是Buffer,对应的实现类PartitionedPairBuffer,设置mapSideCombine=false时会使用该结构;另一种是Map,对应的实现类是PartitionedAppendOnlyMap,设置mapSideCombine=false时会使用该结构。根据是否指定mapSideCombine选项,分别对应不同的处理流程,我们分别说明如下:

  • 设置mapSideCombine=false时

这种情况在Map阶段不进行Combine操作,在内存中缓存记录数据会使用PartitionedPairBuffer这种数据结构来缓存、排序记录数据,它是一个Append-only Buffer,仅支持向Buffer中追加数据键值对记录,PartitionedPairBuffer的结构如下图所示:
PartitionedPairBuffer
默认情况下,PartitionedPairBuffer初始分配的存储容量为capacity = initialCapacity = 64,实际上这个容量是针对key的容量,因为要存储的是键值对记录数据,所以实际存储键值对的容量为2*initialCapacity = 128。PartitionedPairBuffer是一个能够动态扩充容量的Buffer,内部使用一个一维数组来存储键值对,每次扩容结果为当前Buffer容量的2倍,即2*capacity,最大支持存储2^31-1个键值对记录(1073741823个)。
通过上图可以看到,PartitionedPairBuffer存储的键值对记录数据,键是(partition, key)这样一个Tuple,值是对应的数据value,而且curSize是用来跟踪写入Buffer中的记录的,key在Buffer中的索引位置为2*curSize,value的索引位置为2*curSize+1,可见一个键值对的key和value的存储在PartitionedPairBuffer内部的数组中是相邻的。
使用PartitionedPairBuffer缓存键值对记录数据,通过跟踪实际写入到Buffer内的记录数据的字节数来判断,是否需要将Buffer中的数据Spill到磁盘文件,如下代码所示:

  protected def maybeSpill(collection: C, currentMemory: Long): Boolean = {
    var shouldSpill = false
    if (elementsRead % 32 == 0 && currentMemory >= myMemoryThreshold) {
      // Claim up to double our current memory from the shuffle memory pool
      val amountToRequest = 2 * currentMemory - myMemoryThreshold
      val granted = acquireMemory(amountToRequest)
      myMemoryThreshold += granted
      // If we were granted too little memory to grow further (either tryToAcquire returned 0,
      // or we already had more memory than myMemoryThreshold), spill the current collection
      shouldSpill = currentMemory >= myMemoryThreshold
    }
    shouldSpill = shouldSpill || _elementsRead > numElementsForceSpillThreshold
    // Actually spill
    if (shouldSpill) {
      _spillCount += 1
      logSpillage(currentMemory)
      spill(collection)
      _elementsRead = 0
      _memoryBytesSpilled += currentMemory
      releaseMemory()
    }
    shouldSpill
  }

上面elementsRead表示存储到PartitionedPairBuffer中的记录数,currentMemory是对Buffer中的总记录数据大小(字节数)的估算,myMemoryThreshold通过配置项spark.shuffle.spill.initialMemoryThreshold来进行设置的,默认值为5 * 1024 * 1024 = 5M。当满足条件elementsRead % 32 == 0 && currentMemory >= myMemoryThreshold时,会先尝试向MemoryManager申请2 * currentMemory – myMemoryThreshold大小的内存,如果能够申请到,则不进行Spill操作,而是继续向Buffer中存储数据,否则就会调用spill()方法将Buffer中数据输出到磁盘文件。
向PartitionedPairBuffer中写入记录数据,以及满足条件Spill记录数据到磁盘文件,具体处理流程,如下图所示:
ExternalSorter.insertAllWithoutMapSideCombime
为了查看按照怎样的规则进行排序,我们看一下,当不进行Map Side Combine时,创建ExternalSorter对象的代码如下所示:

      // In this case we pass neither an aggregator nor an ordering to the sorter, because we don't
      // care whether the keys get sorted in each partition; that will be done on the reduce side
      // if the operation being run is sortByKey.
      new ExternalSorter[K, V, V](
        context, aggregator = None, Some(dep.partitioner), ordering = None, dep.serializer)

上面aggregator = None,ordering = None,在对PartitionedPairBuffer中的记录数据Spill到磁盘之前,要使用默认的排序规则进行排序,排序的规则是只对PartitionedPairBuffer中的记录按Partition ID进行升序排序,可以查看WritablePartitionedPairCollection伴生对象类的代码(其中PartitionedPairBuffer类实现了特质WritablePartitionedPairCollection),如下所示:

  /**
   * A comparator for (Int, K) pairs that orders them by only their partition ID.
   */
  def partitionComparator[K]: Comparator[(Int, K)] = new Comparator[(Int, K)] {
    override def compare(a: (Int, K), b: (Int, K)): Int = {
      a._1 - b._1
    }
  }

上面图中,引用了SortShuffleWriter.writeBlockFiles这个子序列图,用来生成Block数据文件和索引文件,后面我们会单独说明。通过对RDD进行计算生成一个记录迭代器对象,通过该迭代器迭代出的记录会存储到PartitionedPairBuffer中,当满足Spill条件时,先对PartitionedPairBuffer中记录进行排序,最后Spill到磁盘文件,这个过程中PartitionedPairBuffer中的记录数据的变化情况,如下图所示:
PartitionedPairBuffer-Sorting-and-Spilling
上图中,对内存中PartitionedPairBuffer中的记录按照Partition ID进行排序,并且属于同一个Partition的数据记录在PartitionedPairBuffer内部的data数组中是连续的。排序结束后,在Spill到磁盘文件时,将对应的Partition ID去掉了,只在文件temp_shuffle_4c4b258d-52e4-47a0-a9b6-692f1af7ec9d中连续存储键值对数据,但同时在另一个内存数组结构中会保存文件中每个Partition拥有的记录数,这样就能根据Partition的记录数来顺序读取文件temp_shuffle_4c4b258d-52e4-47a0-a9b6-692f1af7ec9d中属于同一个Partition的全部记录数据。
ExternalSorter类内部维护了一个SpillFile的ArrayBuffer数组,最终可能会生成多个SpillFile,SpillFile的定义如下所示:

  private[this] case class SpilledFile(
    file: File,
    blockId: BlockId,
    serializerBatchSizes: Array[Long],
    elementsPerPartition: Array[Long])

每个SpillFile包含一个blockId,标识Map输出的该临时文件;serializerBatchSizes表示每次批量写入到文件的Object的数量,默认为10000,由配置项spark.shuffle.spill.batchSize来控制;elementsPerPartition表示每个Partition中的Object的数量。调用ExternalSorter的insertAll()方法,最终可能有如下3种情况:

  • Map阶段输出记录数较少,没有生成SpillFile,那么所有数据都在Buffer中,直接对Buffer中记录排序并输出到文件
  • Map阶段输出记录数较多,生成多个SpillFile,同时Buffer中也有部分记录数据
  • Map阶段输出记录数较多,只生成多个SpillFile

有关后续如何对上面3种情况进行处理,可以想见后面对子序列图SortShuffleWriter.writeBlockFiles的说明。

  • 设置mapSideCombine=true时

这种情况在Map阶段会执行Combine操作,在Map阶段进行Combine操作能够降低Map阶段数据记录的总数,从而降低Shuffle过程中数据的跨网络拷贝传输。这时,RDD对应的ShuffleDependency需要设置一个Aggregator用来执行Combine操作,可以看下Aggregator类声明,代码如下所示:

/**
 * :: DeveloperApi ::
 * A set of functions used to aggregate data.
 *
 * @param createCombiner function to create the initial value of the aggregation.
 * @param mergeValue function to merge a new value into the aggregation result.
 * @param mergeCombiners function to merge outputs from multiple mergeValue function.
 */
@DeveloperApi
case class Aggregator[K, V, C] (
    createCombiner: V => C,
    mergeValue: (C, V) => C,
    mergeCombiners: (C, C) => C) {
  ... ...
}

由于在Map阶段只用到了构造Aggregator的几个函数参数createCombiner、mergeValue、mergeCombiners,我们对这几个函数详细说明如下:

  • createCombiner:进行Aggregation开始时,需要设置初始值。因为在Aggregation过程中使用了类似Map的内存数据结构来管理键值对,每次加入前会先查看Map内存结构中是否存在Key对应的Value,第一次肯定不存在,所以首次将某个Key的Value加入到Map内存结构中时,Key在Map内存结构中第一次有了Value。
  • mergeValue:某个Key已经在Map结构中存在Value,后续某次又遇到相同的Key和一个新的Value,这时需要通过该函数,将旧Value和新Value进行合并,根据Key检索能够得到合并后的新Value。
  • mergeCombiners:一个Map内存结构中Key和Value是由mergeValue生成的,那么在向Map中插入数据,肯定会遇到Map使用容量达到上限,这时需要将记录数据Spill到磁盘文件,那么多个Spill输出的磁盘文件中可能存在同一个Key,这时需要对多个Spill输出的磁盘文件中的Key的多个Value进行合并,这时需要使用mergeCombiners函数进行处理。

该类中定义了combineValuesByKey、combineValuesByKey、combineCombinersByKey,由于这些函数是在Reduce阶段使用的,所以在这里先不说明,后续文章我们会单独详细来分析。
我们通过下面的序列图来描述,需要进行Map Side Combine时的处理流程,如下所示:
ExternalSorter.insertAllWithMapSideCombine
对照上图,我们看一下,当需要进行Map Side Combine时,对应的ExternalSorter类insertAll()方法中的处理逻辑,代码如下所示:

    val shouldCombine = aggregator.isDefined

    if (shouldCombine) {
      // Combine values in-memory first using our AppendOnlyMap
      val mergeValue = aggregator.get.mergeValue
      val createCombiner = aggregator.get.createCombiner
      var kv: Product2[K, V] = null
      val update = (hadValue: Boolean, oldValue: C) => {
        if (hadValue) mergeValue(oldValue, kv._2) else createCombiner(kv._2)
      }
      while (records.hasNext) {
        addElementsRead()
        kv = records.next()
        map.changeValue((getPartition(kv._1), kv._1), update)
        maybeSpillCollection(usingMap = true)
      }
    }

上面代码中,map是内存数据结构,最重要的是update函数和map的changeValue方法(这里的map对应的实现类是PartitionedAppendOnlyMap)。update函数所做的工作,其实就是对createCombiner和mergeValue这两个函数的使用,第一次遇到一个Key调用createCombiner函数处理,非首次遇到同一个Key对应新的Value调用mergeValue函数进行合并处理。map的changeValue方法主要是将Key和Value在map中存储或者进行修改(对出现的同一个Key的多个Value进行合并,并将合并后的新Value替换旧Value)。
PartitionedAppendOnlyMap是一个经过优化的哈希表,它支持向map中追加数据,以及修改Key对应的Value,但是不支持删除某个Key及其对应的Value。它能够支持的存储容量是0.7 * 2 ^ 29 = 375809638。当达到指定存储容量或者指定限制,就会将map中记录数据Spill到磁盘文件,这个过程和前面的类似,不再累述。

创建Shuffle Block数据文件及其索引文件

无论是使用PartitionedPairBuffer,还是使用PartitionedAppendOnlyMap,当需要容量满足Spill条件时,都会将该内存结构(buffer/map)中记录数据Spill到磁盘文件,所以Spill到磁盘文件的格式是相同的。对于后续Block数据文件和索引文件的生成逻辑也是相同,如下图所示:
SortShuffleWriter.writeBlockFiles
假设,我们生成的Shuffle Block文件对应各个参数为:shuffleId=2901,mapId=11825,reduceId=0,这里reduceId是一个NOOP_REDUCE_ID,表示与DiskStore进行磁盘I/O交互操作,而DiskStore期望对应一个(map, reduce)对,但是对于排序的Shuffle输出,通常Reducer拉取数据后只生成一个文件(Reduce文件),所以这里默认reduceId为0。经过上图的处理流程,可以生成一个.data文件,也就是Block数据文件;一个.index文件,也就是包含了各个Partition在数据文件中的偏移位置的索引文件。这个过程生成的文件,示例如下所示:

shuffle_2901_11825_0.data
shuffle_2901_11825_0.index

这样,对于每个RDD的多个Partition进行处理后,都会生成对应的数据文件和索引文件,后续在Reduce端就可以读取这些Block文件,这些记录数据在文件中都是经过分区(Partitioned)的。

相关 [spark shuffle 分析] 推荐:

Spark Shuffle过程分析:Map阶段处理流程

- - 简单之美
默认配置情况下,Spark在Shuffle过程中会使用SortShuffleManager来管理Shuffle过程中需要的基本组件,以及对RDD各个Partition数据的计算. 我们可以在Driver和Executor对应的SparkEnv对象创建过程中看到对应的配置,如下代码所示:. 如果需要修改ShuffleManager实现,则只需要修改配置项spark.shuffle.manager即可,默认支持sort和 tungsten-sort,可以指定自己实现的ShuffleManager类.

Spark性能优化——和shuffle搏斗

- - 四火的唠叨
Spark的性能分析和调优很有意思,今天再写一篇. 主要话题是shuffle,当然也牵涉一些其他代码上的小把戏. 以前写过一篇文章,比较了 几种不同场景的性能优化,包括portal的性能优化,web service的性能优化,还有Spark job的性能优化. Spark的性能优化有一些特殊的地方,比如实时性一般不在考虑范围之内,通常我们用Spark来处理的数据,都是要求异步得到结果的数据;再比如数据量一般都很大,要不然也没有必要在集群上操纵这么一个大家伙,等等.

Flume+Spark+Hive+Spark SQL离线分析系统

- - CSDN博客推荐文章
前段时间把Scala和Spark一起学习了,所以借此机会在这里做个总结,顺便和大家一起分享一下目前最火的分布式计算技术Spark. 当然Spark不光是可以做离线计算,还提供了许多功能强大的组件,比如说,Spark Streaming 组件做实时计算,和Kafka等消息系统也有很好的兼容性;Spark Sql,可以让用户通过标准SQL语句操作从不同的数据源中过来的结构化数据;还提供了种类丰富的MLlib库方便用户做机器学习等等.

HADOOP SHUFFLE(转载)

- - 数据库 - ITeye博客
Shuffle过程是MapReduce的核心,也被称为奇迹发生的地方. 要想理解MapReduce,Shuffle是必须要了解的. 我看过很多相关的资料,但每次看完都云里雾里的绕着,很难理清大致的逻辑,反而越搅越混. 前段时间在做MapReduce job性能调优的工作,需要深入代码研究MapReduce的运行机制,这才对Shuffle探了个究竟.

颠覆大数据分析之Spark弹性分布式数据集

- - 并发编程网 - ifeve.com
颠覆大数据分析之Spark弹性数据集. 译者:黄经业     购书. Spark中迭代式机器学习算法的数据流可以通过图2.3来进行理解. 将它和图2.1中Hadoop MR的迭代式机器学习的数据流比较一下. MR中每次迭代都会涉及HDFS的读写,而在Spark中则要简单得多. 它仅需从HDFS到Spark中的分布式共享对象空间的一次读入——从HDFS文件中创建RDD.

[转]MapReduce:详解Shuffle(copy,sort,merge)过程

- - 芒果先生Mango的专栏
Shuffle过程是MapReduce的核心,也被称为奇迹发生的地方. 要想理解MapReduce, Shuffle是必须要了解的. 我看过很多相关的资料,但每次看完都云里雾里的绕着,很难理清大致的逻辑,反而越搅越混. 前段时间在做MapReduce job 性能调优的工作,需要深入代码研究MapReduce的运行机制,这才对Shuffle探了个究竟.

Spark概览

- - 简单文本
Spark具有先进的DAG执行引擎,支持cyclic data flow和内存计算. 因此,它的运行速度,在内存中是Hadoop MapReduce的100倍,在磁盘中是10倍. 这样的性能指标,真的让人心动啊. Spark的API更为简单,提供了80个High Level的操作,可以很好地支持并行应用.

Spark与Mapreduce?

- - 崔永键的博客
我本人是类似Hive平台的系统工程师,我对MapReduce的熟悉程度是一般,它是我的底层框架. 我隔壁组在实验Spark,想将一部分计算迁移到Spark上. 年初的时候,看Spark的评价,几乎一致表示,Spark是小数据集上处理复杂迭代的交互系统,并不擅长大数据集,也没有稳定性. 但是最近的风评已经变化,尤其是14年10月他们完成了Peta sort的实验,这标志着Spark越来越接近替代Hadoop MapReduce了.

Spark迷思

- - ITeye博客
目前在媒体上有很大的关于Apache Spark框架的声音,渐渐的它成为了大数据领域的下一个大的东西. 证明这件事的最简单的方式就是看google的趋势图:. 上图展示的过去两年Hadoop和Spark的趋势. Spark在终端用户之间变得越来越受欢迎,而且这些用户经常在网上找Spark相关资料. 这给了Spark起了很大的宣传作用;同时围绕着它的也有误区和思维错误,而且很多人还把这些误区作为银弹,认为它可以解决他们的问题并提供比Hadoop好100倍的性能.

Spark 优化

- - CSDN博客推荐文章
提到Spark与Hadoop的区别,基本最常说的就是Spark采用基于内存的计算方式,尽管这种方式对数据处理的效率很高,但也会往往引发各种各样的问题,Spark中常见的OOM等等. 效率高的特点,注定了Spark对性能的严苛要求,那Spark不同程序的性能会碰到不同的资源瓶颈,比如:CPU,带宽、内存.