1.Spark企业级应用开发和调优

Spark项目编程优化历程记录，主要介绍了Spark企业级别的开发过程中面临的问题和调优方法。包含合理分配分片，避免计算中间结果（大数据量）的collect,合理使用map,优化广播变量等操作，降低网络和磁盘IO，提高计算效率。

2.核心技术优化方法对比

首先如下图（2.1），Spark应用开发在集群（伪分布式）中的记录,每一种不同颜色的折线代表一个分布式机器

最终,图4中四条折线并行达到峰值(即CPU100%).降低了处理时间,增大了处理效率.

2.1.重要并行计算模型构建对比

图1 传统方式计算模型在模拟集群计算概览图

图2 spark并行模型1在模拟集群并行计算概览图

图3 spark并行模型2在模拟集群并行计算概览图

图4 spark并行模型3在模拟集群并行计算概览图

2.2.Spark优化技术要点

2.2.1.如何构建一个合理的弹性分布式数据集(RDD)

Spark之所以快速,一是分而治之,二是允许基于内存计算.
第一步,常用的构建一个分布式数据方式:

• 方式一:基于文件读取
  
  • textFile(name, minPartitions=None, use_unicode=True) 返回RDD,可以读取text本地文件,HDFS等等

sc.textFile("file:///native/dir")
sc.textFile("/HDFS/dir")

• 方式二:基于内存读取
  
  • parallelize(c, numSlices=None) 返回RDD,基于内存读取.

sc.parallelize([0, 2, 3, 4, 6], 5).glom().collect()

在项目模型中,计算模型将的x,y坐标xyload = sc.parallelize(xyload)通过内存读成RDD模式.

2.2.2.如何处理一个弹性分布式数据集

在处理弹性是分布式数据集之前,应该充分利用RDD本质的性质,RDD执行策略是懒操作,在转换和执行两个状态中,只有执行才会真正去计算,如将一个文件textFile至RDD,这个文件并没有做物理上的动作,而RDD只是逻辑映射,当执行college或者split等可以返回一个新RDD时,才会发生资源分配,计算.可以简单理解为,一个RDD转变成另个新的RDD时,才发生了真正的资源调度,计算,IO等操作.

在项目中,

cellist=xyload.map(getCellList)
cellisttxt = cellist.filter(lambda x : x != None)

其中,

• map

map(f, preservesPartitioning=False)

返回一个新的RDD,并对RDD中的每个元素做操作(如功能函数的运算或者定义的循环,针对的元素级别的)
在项目中,实现Celllist循环操作,操作级别对弹性分布式元素中的每个元素.

• filter(f)

cellisttxt = cellist.filter(lambda x : x != None)

返回一个新的RDD,包含满足功能函数的元素.
在项目中,实现返回cellist中元素去除None元素,保证RDD后续业务操作正确性.

2.2.3.如何优化处理数据过大的中间结果

RDD的collect操作可以实现元素级别的聚合,但是这个执行过程会造成单一driver大量IO,内存占用过大,网络传输量大等等瓶颈.

所以,在getcellist方法后,将分布式持久化,然后再通过文件批量依次读取过程,避开driver开销过大的难题.

2.2.4.广播变量的合理使用

增加广播变量降低读写。适用于某变量需要反复使用，如在各个分片中都有一个数组固定的计算值，这个数组不要反复从文件读取而直接用广播变量，最大限度降低集群的IO.

3.大数据模型开发历程

由图5,在企业中开发Spark应用，以接口的服务方式,第一次post大数据平台文件上传服务,上传所需的数据文件,二次post调用服务接口,传入Spark分布式模型必备的参数,包括执行本次执行ID,输入路径,输出路径.一期模型开始监控大数据平台执行返回的状态.

此时,基于每个RDD内存做计算,map操作得到getcellist,并通过filter去除脏数据(None),形成中间结果,分布式持久化,最后通过numpy依次读取持久化文件,并做排序后保存成最终结果.

图5 分布式Spark模型的主要执行过程示意图

最终,业务平台通过大数据平台监控得到执行成功状态信号,get最终结果文件至业务平台.