你好,我是吴磊。
上一讲我们说,想要提升CPU利用率,最重要的就是合理分配执行内存,但是,执行内存只是Spark内存分区的一部分。因此,想要合理分配执行内存,我们必须先从整体上合理划分好Spark所有的内存区域。
可在实际开发应用的时候,身边有不少同学向我抱怨:“Spark划分不同内存区域的原理我都知道,但我还是不知道不同内存区域的大小该怎么设置,纠结来、纠结去。最后,所有跟内存有关的配置项,我还是保留了默认值。”
这种不能把原理和实践结合起来的情况很常见,所以今天这一讲,我就从熟悉的Label Encoding实例出发,一步步带你去分析不同情况下,不同内存区域的调整办法,帮你归纳出最大化内存利用率的常规步骤。这样,你在调整内存的时候,就能结合应用的需要,做到有章可循、有的放矢。
我们先来回顾一下第5讲中讲过的Label Encoding。在Label Encoding的业务场景中,我们需要对用户兴趣特征做Encoding。依据模板中兴趣字符串及其索引位置,我们的任务是把千亿条样本中的用户兴趣转换为对应的索引值。模板文件的内容示例如下所示。
//模板文件
//用户兴趣
体育-篮球-NBA-湖人
军事-武器-步枪-AK47
实现的代码如下所示,注意啦,这里的代码是第5讲中优化后的版本。
/**
输入参数:模板文件路径,用户兴趣字符串
返回值:用户兴趣字符串对应的索引值
*/
//函数定义
val findIndex: (String) => (String) => Int = {
(filePath) =>
val source = Source.fromFile(filePath, "UTF-8")
val lines = source.getLines().toArray
source.close()
val searchMap = lines.zip(0 until lines.size).toMap
(interest) => searchMap.getOrElse(interest, -1)
}
val partFunc = findIndex(filePath)
//Dataset中的函数调用
partFunc("体育-篮球-NBA-湖人")
下面,咱们先一起回顾一下代码实现思路,再来分析它目前存在的性能隐患,最后去探讨优化它的方法。
首先,findIndex函数的主体逻辑比较简单,就是读取模板文件和构建Map映射,以及查找用户兴趣并返回索引。不过,findIndex函数被定义成了高阶函数。这样一来,当以模板文件为实参调用这个高阶函数的时候,我们会得到一个内置了Map查找字典的标量函数partFunc,最后在千亿样本上调用partFunc完成数据转换。利用高阶函数,我们就避免了让Executor中的每一个Task去读取模板文件,以及从头构建Map字典这种执行低效的做法。
在运行时,这个函数在Driver端会被封装到一个又一个的Task中去,随后Driver把这些Task分发到Executor,Executor接收到任务之后,交由线程池去执行(调度系统的内容可以回顾第5讲)。这个时候,每个Task就像是一架架小飞机,携带着代码“乘客”和数据“行李”,从Driver飞往Executor。Task小飞机在Executor机场着陆之后,代码“乘客”乘坐出租车或是机场大巴,去往JVM stack;数据“行李”则由专人堆放在JVM Heap,也就是我们常说的堆内内存。
回顾Label encoding中的findIndex函数不难发现,其中大部分都是代码“乘客”,唯一的数据“行李”是名为searchMap的Map字典。像这样用户自定义的数据结构,消耗的内存区域就是堆内内存的User Memory(Spark对内存区域的划分内容可以回顾一下第7讲)。
回顾到这里,你觉得findIndex函数有没有性能隐患呢?你可以先自己思考一下,有了答案之后再来看我下面的分析。
答案当然是“有”。首先,每架小飞机都携带这么一份数据“大件行李”,自然需要消耗更多的“燃油”,这里的“燃油”指的是Task分发过程中带来的网络开销。其次,因为每架小飞机着陆之后,都会在Executor的“旅客行李专区”User Memory寄存上这份同样的数据“行李”,所以,User Memory需要确保有足够的空间可以寄存所有旅客的行李,也就是大量的重复数据。
那么,User Memory到底需要准备出多大的内存空间才行呢?我们不妨来算一算。这样的计算并不难,只需要用飞机架次乘以行李大小就可以了。
用户自定义的数据结构往往是用于辅助函数完成计算任务的,所以函数执行完毕之后,它携带的数据结构的生命周期也就告一段落。因此,在Task的数量统计上,我们不必在意一个Executor总共需要处理多少个Task,只需要关注它在同一时间可以并行处理的Task数量,也就是Executor的线程池大小即可。
我们说过,Executor线程池大小由spark.executor.cores和spark.task.cpus这两个参数的商(spark.executor.cores/spark.task.cpus)决定,我们暂且把这个商记作#threads。
接下来是估算数据“行李”大小,由于searchMap并不是分布式数据集,因此我们不必采用先Cache,再提取Spark执行计划统计信息的方式。对于这样的Java数据结构,我们完全可以在REPL中,通过Java的常规方法估算数据存储大小,估算得到的searchMap大小记为#size。
好啦!现在,我们可以算出,User Memory至少需要提供#threads * #size这么大的内存空间,才能支持分布式任务完成计算。但是,对于User Memory内存区域来说,使用#threads * #size的空间去重复存储同样的数据,本身就是降低了内存的利用率。那么,我们该怎么省掉#threads * #size的内存消耗呢?
学习过广播变量之后,想必你头脑中已经有了思路。没错,咱们可以尝试使用广播变量,来对示例中的代码进行优化。
仔细观察findIndex函数,我们不难发现,函数的核心计算逻辑有两点。一是读取模板文件、创建Map映射字典;二是以给定字符串对字典进行查找,并返回查找结果。显然,千亿样本转换的核心需求是其中的第二个环节。既然如此,我们完全可以把创建好的Map字典封装成广播变量,然后分发到各个Executors中去。
有了广播变量的帮忙,凡是发往同一个Executor的Task小飞机,都无需亲自携带数据“行李”,这些大件行李会由“联邦广播快递公司”派货机专门发往各个Executors,Driver和每个Executors之间,都有一班这样的货运专线。思路说完了,优化后的代码如下所示。
/**
广播变量实现方式
*/
//定义广播变量
val source = Source.fromFile(filePath, "UTF-8")
val lines = source.getLines().toArray
source.close()
val searchMap = lines.zip(0 until lines.size).toMap
val bcSearchMap = sparkSession.sparkContext.broadcast(searchMap)
//在Dataset中访问广播变量
bcSearchMap.value.getOrElse("体育-篮球-NBA-湖人", -1)
上面代码的实现思路很简单:第一步还是读取模板文件、创建Map字典;第二步,把Map字典封装为广播变量。这样一来,在对千亿样本进行转换时,我们直接通过bcSearchMap.value读取广播变量内容,然后,通过调用Map字典的getOrElse方法来获取用户兴趣对应的索引值。
相比最开始的第一种实现方式,第二种实现方式的代码改动还是比较小的,那这一版代码对内存的消耗情况有什么改进呢?
我们发现,Task小飞机的代码“乘客”换人了!小飞机之前需要携带函数findIndex,现在则换成了一位“匿名的乘客”:一个读取广播变量并调用其getOrElse方法的匿名函数。由于这位“匿名的乘客”将大件行李托运给了“联邦广播快递公司”的专用货机,因此,Task小飞机着陆后,没有任何“行李”需要寄存到User Memory。换句话说,优化后的版本不会对User Memory内存区域进行占用,所以第一种实现方式中#threads * #size的内存消耗就可以省掉了。
这样一来,原来的内存消耗转嫁到了广播变量身上。但是,广播变量也会消耗内存,这会不会带来新的性能隐患呢?那我们就来看看,广播变量消耗的具体是哪块内存区域。
回顾存储系统那一讲,我们说过,Spark存储系统主要有3个服务对象,分别是Shuffle中间文件、RDD缓存和广播变量。它们都由Executor上的BlockManager进行管理,对于数据在内存和磁盘中的存储,BlockManager利用MemoryStore和DiskStore进行抽象和封装。
那么,广播变量所携带的数据内容会物化到MemoryStore中去,以Executor为粒度为所有Task提供唯一的一份数据拷贝。MemoryStore产生的内存占用会被记入到Storage Memory的账上。因此,广播变量消耗的就是Storage Memory内存区域。
接下来,我们再来盘算一下,第二种实现方式究竟需要耗费多少内存空间。由于广播变量的分发和存储以Executor为粒度,因此每个Executor消耗的内存空间,就是searchMap一份数据拷贝的大小。searchMap的大小我们刚刚计算过就是#size。
明确了Storage Memory内存区域的具体消耗之后,我们自然可以根据公式:(spark.executor.memory – 300MB)* spark.memory.fraction * spark.memory.storageFraction去有针对性地调节相关的内存配置项。
现在,咱们在两份不同的代码实现下,分别定量分析了不同内存区域的消耗与占用。对于这些消耗做到心中有数,我们自然就能够相应地去调整相关的配置项参数。基于这样的思路,想要最大化内存利用率,我们需要遵循两个步骤:
我们以堆内内存为例,来讲一讲内存规划的两步走具体该如何操作。我们都知道,堆内内存划分为Reserved Memory、User Memory、Storage Memory和Execution Memory这4个区域。预留内存固定为300MB,不用理会,其他3个区域需要你去规划。
首先,我们来说内存占用的预估,主要分为三步。
第一步,计算User Memory的内存消耗。我们先汇总应用中包含的自定义数据结构,并估算这些对象的总大小#size,然后用#size乘以Executor的线程池大小,即可得到User Memory区域的内存消耗#User。
第二步,计算Storage Memory的内存消耗。我们先汇总应用中涉及的广播变量和分布式数据集缓存,分别估算这两类对象的总大小,分别记为#bc、#cache。另外,我们把集群中的Executors总数记作#E。这样,每个Executor中Storage Memory区域的内存消耗的公式就是:#Storage = #bc + #cache / #E。
第三步,计算执行内存的消耗。学习上一讲,我们知道执行内存的消耗与多个因素有关。第一个因素是Executor线程池大小#threads,第二个因素是数据分片大小,而数据分片大小取决于数据集尺寸#dataset和并行度#N。因此,每个Executor中执行内存的消耗的计算公式为:#Execution = #threads * #dataset / #N。
得到这3个内存区域的预估大小#User、#Storage、#Execution之后,调整相关的内存配置项就是一件水到渠成的事情(由公式(spark.executor.memory – 300MB)* spark.memory.fraction * spark.memory.storageFraction)可知),这里我们也可以分为3步。
首先,根据定义,spark.memory.fraction可以由公式(#Storage + #Execution)/(#User + #Storage + #Execution)计算得到。
同理,spark.memory.storageFraction的数值应该参考(#Storage)/(#Storage + #Execution)。
最后,对于Executor堆内内存总大小spark.executor.memory的设置,我们自然要参考4个内存区域的总消耗,也就是300MB + #User + #Storage + #Execution。不过,我们要注意,利用这个公式计算的前提是,不同内存区域的占比与不同类型的数据消耗一致。
总的来说,在内存规划的两步走中,第一步预估不同区域的内存占比尤为关键,因为第二步中参数的调整完全取决于第一步的预估结果。如果你按照这两个步骤去设置相关的内存配置项,相信你的应用在运行时就能够充分利用不同的内存区域,避免出现因参数设置不当而导致的内存浪费现象,从而在整体上提升内存利用率。
合理划分Spark所有的内存区域,是同时提升CPU与内存利用率的基础。因此,掌握内存规划很重要,在今天这一讲,我们把内存规划归纳为两步走。
第一步是预估内存占用。
第二步是调整内存配置项:根据公式得到的3个内存区域的预估大小#User、#Storage、#Execution,去调整(spark.executor.memory – 300MB)* spark.memory.fraction * spark.memory.storageFraction公式中涉及的所有配置项。
这里,我还想多说几句,内存规划两步走终归只是手段,它最终要达到的效果和目的,是确保不同内存区域的占比与不同类型的数据消耗保持一致,从而实现内存利用率的最大化。
期待在留言区看到你的思考和答案,我们下一讲见!