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也只需将满足条件的小表完整加载到内存。

相关文章
|
1月前
|
存储 分布式计算 Java
|
1月前
|
存储 编译器 C语言
【C语言篇】数据在内存中的存储(超详细)
浮点数就采⽤下⾯的规则表⽰,即指数E的真实值加上127(或1023),再将有效数字M去掉整数部分的1。
|
1月前
|
编译器 C++
virtual类的使用方法问题之C++类中的非静态数据成员是进行内存对齐的如何解决
virtual类的使用方法问题之C++类中的非静态数据成员是进行内存对齐的如何解决
|
1月前
|
存储 算法 大数据
小米教你:2GB内存搞定20亿数据的高效算法
你好,我是小米。本文介绍如何在2GB内存中找出20亿个整数里出现次数最多的数。通过将数据用哈希函数分至16个小文件,每份独立计数后选出频次最高的数,最终比对得出结果。这种方法有效解决大数据下的内存限制问题,并可应用于更广泛的场景。欢迎关注我的公众号“软件求生”,获取更多技术分享!
147 12
|
1月前
|
分布式计算 监控 大数据
如何处理 Spark 中的倾斜数据?
【8月更文挑战第13天】
177 4
|
1月前
|
存储 缓存 分布式计算
|
1月前
|
SQL 存储 分布式计算
|
1月前
|
分布式计算 Apache 数据安全/隐私保护
流计算引擎数据问题之在 Spark Structured Streaming 中水印计算和使用如何解决
流计算引擎数据问题之在 Spark Structured Streaming 中水印计算和使用如何解决
38 1
|
1月前
|
C++ 容器
curl使用小记(三)——获取远端数据到内存缓冲区
curl使用小记(三)——获取远端数据到内存缓冲区
35 0
|
1月前
|
存储 编译器 C语言
数据在内存中的存储
数据在内存中的存储