Spark在处理数据的时候,会将数据都加载到内存再做处理吗?-阿里云开发者社区

开发者社区> 大数据> 正文

Spark在处理数据的时候,会将数据都加载到内存再做处理吗?

简介: 对于Spark的初学者,往往会有一个疑问:Spark(如SparkRDD、SparkSQL)在处理数据的时候,会将数据都加载到内存再做处理吗?

对于Spark的初学者,往往会有一个疑问:Spark(如SparkRDD、SparkSQL)在处理数据的时候,会将数据都加载到内存再做处理吗?

很显然,答案是否定的!

对该问题产生疑问的根源还是对Spark计算模型理解不透彻。

对于Spark RDD,它是一个分布式的弹性数据集,不真正存储数据。如果你没有在代码中调用persist或者cache算子,Spark是不会真正将数据都放到内存里的。

此外,还要考虑persist/cache的缓存级别,以及对什么进行缓存(比如是对整张表生成的DataSet缓存还是列裁剪之后生成的DataSet缓存)(关于Spark RDD的特性解析参考《Spark RDD详解》

既然Spark RDD不存储数据,那么它内部是如何读取数据的呢?其实Spark内部也实现了一套存储系统:BlockManager。为了更深刻的理解Spark RDD数据的处理流程,先抛开BlockManager本身原理,从源码角度阐述RDD内部函数的迭代体系。

我们都知道RDD算子最终会被转化为shuffle map task和result task,这些task通过调用RDD的iterator方法获取对应partition数据,而这个iterator方法又会逐层调用父RDD的iterator方法获取数据(通过重写scala.collection.iterator的hasNext和next方法实现)。主要过程如下:

首先看ShuffleMapTask和ResultTask中runTask方法的源码:

1.jpg
2.jpg

关键看这部分处理逻辑:

rdd.iterator(partition, context)

3.jpg

getOrCompute方法会先通过当前executor上的BlockManager获取指定blockId的block,如果block不存在则调用computeOrReadCheckpoint,如果要处理的RDD没有被checkpoint或者materialized,则接着调用compute方法进行计算。

compute方法是RDD的抽象方法,由继承RDD的子类具体实现。

以WordCount为例:

sc.textFile(input)
  .flatMap(line => line.split(" "))
  .map(word => (word, 1))
  .reduceByKey(_ + _)
  .saveAsTextFile(output)

1.textFile会构建一个HadoopRDD

2.flatMap/map会构建一个MapPartitionsRDD

3.reduceByKey触发shuffle时会构建一个ShuffledRDD

4.saveAsTextFile作为action算子会触发整个任务的执行

以flatMap/map产生的MapPartitionsRDD实现的compute方法为例:

override def compute(split: Partition, context: TaskContext): Iterator[U] =
    f(context, split.index, firstParent[T].iterator(split, context))

底层调用了parent RDD的iterator方法,然后作为参数传入到了当前的MapPartitionsRDD。而f函数就是对parent RDD的iterator调用了相同的map类函数以执行用户给定的函数。

所以,这是一个逐层嵌套的rdd.iterator方法调用,子RDD调用父RDD的iterator方法并在其结果之上调用Iterator的map函数以执行用户给定的函数,逐层调用直到调用到最初的iterator(比如上述WordCount示例中HadoopRDD partition的iterator)。

而scala.collection.Iterator的map/flatMap方法返回的Iterator就是基于当前Iterator重写了next和hasNext方法的Iterator实例。比如,对于map函数,结果Iterator的hasNext就是直接调用了self iterator的hasNext,next方法就是在self iterator的next方法的结果上调用了指定的map函数。

flatMap和filter函数稍微复杂些,但本质上一样,都是通过调用self iterator的hasNext和next方法对数据进行遍历和处理。

所以,当我们调用最终结果iterator的hasNext和next方法进行遍历时,每遍历一个数据元素都会逐层调用父层iterator的hasNext和next方法。各层的map函数组成一个pipeline,每个数据元素都经过这个pipeline的处理得到最终结果。

这也是Spark的优势之一,map类算子整个形成类似流式处理的pipeline管道,一条数据被该链条上的各个RDD所包裹的函数处理。

再回到WordCount例子。HadoopRDD直接跟数据源关联,内存中存储多少数据跟读取文件的buffer和该RDD的分区数相关(比如buffer*partitionNum,当然这是一个理论值),saveAsTextFile与此类似。MapPartitionsRDD里实际在内存里的数据也跟partition数有关系。ShuffledRDD稍微复杂些,因为牵扯到shuffle,但是RDD本身的特性仍然满足(记录文件的存储位置)。

说完了Spark RDD,再来看另一个问题:Spark SQL对于多表之间join操作,会先把所有表中数据加载到内存再做处理吗?

当然,肯定也不需要!

具体可以查看Spark SQL针对相应的Join SQL的查询计划,以及在之前的文章《Spark SQL如何选择join策略》中,针对目前Spark SQL支持的join方式,任何一种都不要将join语句中涉及的表全部加载到内存。即使是Broadcast Hash Join也只需将满足条件的小表完整加载到内存。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
大数据
使用钉钉扫一扫加入圈子
+ 订阅

大数据计算实践乐园,近距离学习前沿技术

其他文章