大数据中的Spark指的是什么？

Spark是一种通用的大数据计算框架，和传统的大数据技术MapReduce有本质区别。前者是基于内存并行计算的框架，而mapreduce侧重磁盘计算。Spark是加州大学伯克利分校AMP实验室开发的通用内存并行计算框架，用于构建大型的、低延迟的数据分析应用程序。

Spark同样支持离线计算和实时计算两种模式。Spark离线计算速度要比Mapreduce快10-100倍。而实时计算方面，则依赖于SparkStreaming的批处理能力，吞吐量大。不过相比Storm，SparkStreaming并不能做到真正的实时。

Spark使用强大的函数式语言Scala开发，方便简单。同时，它还提供了对Python、Java和R语言的支持。

作为大数据计算框架MapReduce的继任者，Spark具备以下优势特性。

1，高效性

不同于MapReduce将中间计算结果放入磁盘中，Spark采用内存存储中间计算结果，减少了迭代运算的磁盘IO，并通过并行计算DAG图的优化，减少了不同任务之间的依赖，降低了延迟等待时间。内存计算下，Spark 比 MapReduce 快100倍。

2，易用性

不同于MapReduce仅支持Map和Reduce两种编程算子，Spark提供了超过80种不同的Transformation和Action算子，如map,reduce,filter,groupByKey,sortByKey,foreach等，并且采用函数式编程风格，实现相同的功能需要的代码量极大缩小。

3，通用性

Spark提供了统一的解决方案。Spark可以用于批处理、交互式查询（Spark SQL）、实时流处理（Spark Streaming）、机器学习（Spark MLlib）和图计算（GraphX）。

4，兼容性

Spark能够跟很多开源工程兼容使用。如Spark可以使用Hadoop的YARN和Apache Mesos作为它的资源管理和调度器，并且Spark可以读取多种数据源，如HDFS、HBase、MySQL等。

大概4.5个小时

Apache Spark 是专为大规模数据处理而设计的快速通用的计算引擎。Spark是UC Berkeley AMP lab (加州大学伯克利分校的AMP实验室)所开源的类Hadoop MapReduce的通用并行框架，Spark，拥有Hadoop MapReduce所具有的优点；但不同于MapReduce的是——Job中间输出结果可以保存在内存中，从而不再需要读写HDFS，因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。

Spark 是一种与 Hadoop 相似的开源集群计算环境，但是两者之间还存在一些不同之处，这些有用的不同之处使 Spark 在某些工作负载方面表现得更加优越，换句话说，Spark 启用了内存分布数据集，除了能够提供交互式查询外，它还可以优化迭代工作负载。

写这篇博客的起因我在跑一个spark job时，有时能跑通，有时跑不通。程序的需求是对比两个hbase表。程序逻辑很简单，分别从两个hbase表读取全量数据，然后以cogroup二者，对比同一个rowkey下每个列是否一致。

跑不通的错误日志如下：

17/02/25 21:24:20 INFO collection.ExternalAppendOnlyMap: Thread 1896 spilling in-memory map of 83.6 MB to disk (46 times so far)

17/02/25 21:24:22 WARN server.TransportChannelHandler: Exception in connection from /10.110.1.57:57832

java.io.IOException: Connection reset by peer

at sun.nio.ch.FileDispatcherImpl.read0(Native Method)

……

17/02/25 21:24:22 ERROR server.TransportRequestHandler: Error sending result ChunkFetchSuccess{streamChunkId=StreamChunkId{streamId=1258210057016, chunkIndex=0}, buffer=FileSegmentManagedBuffer{file=/data-g/hadoop/yarn/local-dir/usercache/test/appcache/application_1466564207556_1562806/blockmgr-ebe23f0d-5a9e-4a37-952b-73bfab6cceed/3f/shuffle_0_6_0.data, offset=474965639, length=95049579}} to /10.130.1.27:53263closing connection

java.nio.channels.ClosedChannelException

17/02/25 21:24:22 ERROR client.TransportResponseHandler: Still have 1 requests outstanding when connection from c1-hd-dn8.bdp.idc/10.130.1.27:50014 is closed

17/02/25 21:24:22 INFO shuffle.RetryingBlockFetcher: Retrying fetch (1/3) for 1 outstanding blocks after 5000 ms

17/02/25 21:24:22 INFO collection.ExternalAppendOnlyMap: Thread 65 spilling in-memory map of 84.1 MB to disk (44 times so far)

17/02/25 21:24:23 INFO collection.ExternalAppendOnlyMap: Thread 1895 spilling in-memory map of 83.9 MB to disk (47 times so far)

 17/02/25 21:24:27 ERROR shuffle.RetryingBlockFetcher: Exception while beginning fetch of 1 outstanding blocks (after 1 retries)

java.io.IOException: Failed to connect to someHost/someIP:50004

……

Caused by: java.net.ConnectException: Connection refused: someHost/someIP:50004

……

针对上面这个问题，做了相关的尝试，解决了以下几个问题：

（1）运行spark job该分配多少资源，即我们该分配多少个executor？每个executor分配多少内存、多少个core？

（2）spark job 的并行度由什么因素决定？

（3）为什么yarn UI也的executor显示的used memory内存大小比配置的内存小？

该分配多少资源主要看输入量的大小、资源计算的复杂度。一般瓶颈会在shuffle阶段，如果执行某个shuffle的task内存不足，那很可能会跑不下去，程序挂掉。

spark中的计算任务都是一个个task单独执行，executor内存越多，单个task执行时内存越足，执行越顺利。 executor越多，core越多，可并行执行的task数目也就越多。假如总共100个task，5个executor，4个core，那么平均需要执行100/(5*4) = 5个批次；如果是2个executor，4个core,那么需要执行100/(2*4) = 13个批次。

core的数量一般根据内存大小和机器物理核数来定。最好不要超过物理核数。如果executor内存是4G，分配了4个core，那么每个core只有4G/4 = 1G内存。所以core不宜太大，如果太大，每个task执行时的内存将会变小，影响正常执行。

举个例子，我们的输入是两个hbase表，均为3.5G。shuffle阶段两个表会根据rowkey 做join,会产生几十G的shuffle数据。我们这样设置资源：

--driver-memory 1g \

--executor-memory 4g \

--num-executors 6 \

--executor-cores 4 \

并行度分为理论上最大的并行度和实际执行的并行度两种，“理论上”指的是总共的partition数目，一个partition对应一个task执行，如果数据有100个partition，那么理论上并行度最高可以达到100。“实际执行”指的是这些task实际分到executor各个core执行时的并行度。加入有100个partition，但是分配的资源只有10个executor，每个executor2个core，那么他们的并行度是10*2=40， 实际执行时会分批执行，分为100/(10*2) = 5批。我们一般讨论的并行度是理论上的并行度。

并行度（partition数目）由初始数据大小、初始数据类型，程序中设定的numPartitions大小，分配资源的executor、core数目共同决定。 并行度一般在shuffle时发生改变，如果未设定，则默认取上一个stage中最大的partition数目作为当前stage的并行度。所以如果不做设定，那么并行度与初始数据的并行度紧密相关。

1.初始数据文件类型因素

如果读入的数据为hdfs文件，那么默认的并行度是block数量。block大小默认是64MB或128MB。

如果读入的数据是hbase表，那么默认的并行度是表的region数目。

2.人为设定numPartitions

如果人为的在读取数据或者在shuffle类算子中设定numPartitions，那么整体的并行度将会以人为设定的为准。

3.人为设定spark.default.parallelism

spark.default.parallelism参数是全局的，优先级低于人为设定的numPartiton。在shuffle时，如果没有设定numPartiton，那么将为以spark.default.parallelism设定的数目作为并行度。

4.系统默认的spark.default.parallelism

系统默认的spark.default.parallelism = executor数目*core数目

以上4个因素的优先级：

1.numPartitions参数 >2. spark.default.parallelism参数 >3. 读取初始文件产生的并行度

spark中的内存分为多个部分，UI页面上显示的只是缓存RDD用的storage memory，大约是（总内存 - 300M) * 60% * 50% 的量，所以会偏小。具体内存分配如下图：

以上。

---