轉(zhuǎn)載本文需注明出處：微信公眾號(hào)EAWorld，違者必究。

引言：

Spark是在借鑒了MapReduce之上發(fā)展而來(lái)的，繼承了其分布式并行計(jì)算的優(yōu)點(diǎn)并改進(jìn)了MapReduce明顯的缺陷。Spark主要包含了Spark Core、Spark SQL、Spark Streaming、MLLib和GraphX等組件。

本文主要分析了 Spark RDD 以及 RDD 作為開(kāi)發(fā)的不足之處，介紹了 SparkSQL 對(duì)已有的常見(jiàn)數(shù)據(jù)系統(tǒng)的操作方法，以及重點(diǎn)介紹了普元在眾多數(shù)據(jù)開(kāi)發(fā)項(xiàng)目中總結(jié)的基于 SparkSQL Flow 開(kāi)發(fā)框架。

目錄：

一、Spark RDD

二、基于Spark RDD數(shù)據(jù)開(kāi)發(fā)的不足

三、SparkSQL

四、SparkSQL Flow

一、Spark RDD

RDD（Resilient Distributed Dataset）叫做彈性分布式數(shù)據(jù)集，是Spark中最基本的數(shù)據(jù)抽象，它代表一個(gè)不可變、可分區(qū)、元素可并行計(jì)算的集合。

RDD具有數(shù)據(jù)流模型的特點(diǎn)：自動(dòng)容錯(cuò)、位置感知性調(diào)度和可伸縮性。

／／Scala 在內(nèi)存中使用列表創(chuàng)建

val lines ＝ List（“A”， “B”， “C”， “D” …）val rdd：RDD ＝ sc．parallelize（lines）；

可左右滑動(dòng)查看代碼

／／以文本文件創(chuàng)建

val rdd：RDD［String］ ＝ sc．textFile（“hdfs：／／path／filename”）

可左右滑動(dòng)查看代碼

Spark RDD Partition 分區(qū)劃分

新版本的 Hadoop 已經(jīng)把 BlockSize 改為 128M，也就是說(shuō)每個(gè)分區(qū)處理的數(shù)據(jù)量更大。

Spark 讀取文件分區(qū)的核心原理

本質(zhì)上，Spark 是利用了 Hadoop 的底層對(duì)數(shù)據(jù)進(jìn)行分區(qū)的 API（InputFormat）：

public abstract class InputFormat＜K，V＞｛  public abstract List＜InputSplit＞ getSplits（JobContextcontext                               ） throwsIOException，InterruptedException；    public abstract RecordReader＜K，V＞ createRecordReader（InputSplitsplit，                                         TaskAttemptContextcontext                                        ）throwsIOException，InterruptedException；｝

可左右滑動(dòng)查看代碼

Spark 任務(wù)提交后通過(guò)對(duì)輸入進(jìn)行 Split，在 RDD 構(gòu)造階段，只是判斷是否可 Split（如果參數(shù)異常一定在此階段報(bào)出異常），并且 Split 后每個(gè) InputSplit 都是一個(gè)分區(qū)。只有在Action 算子提交后，才真正用 getSplits 返回的 InputSplit 通過(guò) createRecordReader 獲得每個(gè) Partition 的連接。

然后通過(guò) RecordReader 的 next（） 遍歷分區(qū)內(nèi)的數(shù)據(jù)。

Spark RDD 轉(zhuǎn)換函數(shù)和提交函數(shù)

Spark RDD 的眾多函數(shù)可分為兩大類Transformation 與 Action。Transformation 與 Action 的區(qū)別在于，對(duì) RDD 進(jìn)行 Transformation 并不會(huì)觸發(fā)計(jì)算：Transformation 方法所產(chǎn)生的 RDD 對(duì)象只會(huì)記錄住該 RDD 所依賴的 RDD 以及計(jì)算產(chǎn)生該 RDD 的數(shù)據(jù)的方式；只有在用戶進(jìn)行 Action 操作時(shí)，Spark 才會(huì)調(diào)度 RDD 計(jì)算任務(wù)，依次為各個(gè) RDD 計(jì)算數(shù)據(jù)。這就是 Spark RDD 內(nèi)函數(shù)的“懶加載”特性。

二、基于Spark RDD數(shù)據(jù)開(kāi)發(fā)的不足

由于MapReduce的shuffle過(guò)程需寫(xiě)磁盤(pán)，比較影響性能；而Spark利用RDD技術(shù)，計(jì)算在內(nèi)存中流式進(jìn)行。另外 MapReduce計(jì)算框架（API）比較局限， 使用需要關(guān)注的參數(shù)眾多，而Spark則是中間結(jié)果自動(dòng)推斷，通過(guò)對(duì)數(shù)據(jù)集上鏈?zhǔn)綀?zhí)行函數(shù)具備一定的靈活性。

即使 SparkRDD 相對(duì)于 MapReduce 提高很大的便利性，但在使用上仍然有許多問(wèn)題。體現(xiàn)在一下幾個(gè)方面：

RDD 函數(shù)眾多，開(kāi)發(fā)者不容易掌握，部分函數(shù)使用不當(dāng) shuffle時(shí)造成數(shù)據(jù)傾斜影響性能；

RDD 關(guān)注點(diǎn)仍然是Spark太底層的 API，基于 Spark RDD的開(kāi)發(fā)是基于特定語(yǔ)言（Scala，Python，Java）的函數(shù)開(kāi)發(fā)，無(wú)法以數(shù)據(jù)的視界來(lái)開(kāi)發(fā)數(shù)據(jù)；

對(duì) RDD 轉(zhuǎn)換算子函數(shù)內(nèi)部分常量、變量、廣播變量使用不當(dāng)，會(huì)造成不可控的異常；

對(duì)多種數(shù)據(jù)開(kāi)發(fā)，需各自開(kāi)發(fā)RDD的轉(zhuǎn)換，樣板代碼較多，無(wú)法有效重利用；

其它在運(yùn)行期可能發(fā)生的異常。如：對(duì)象無(wú)法序列化等運(yùn)行期才能發(fā)現(xiàn)的異常。