怎么看懂Spark的基本原理

一，Spark優(yōu)勢(shì)特點(diǎn)

作為大數(shù)據(jù)計(jì)算框架MapReduce的繼任者，Spark具備以下優(yōu)勢(shì)特性。

1，高效性

不同于MapReduce將中間計(jì)算結(jié)果放入磁盤中，Spark采用內(nèi)存存儲(chǔ)中間計(jì)算結(jié)果，減少了迭代運(yùn)算的磁盤IO，并通過并行計(jì)算DAG圖的優(yōu)化，減少了不同任務(wù)之間的依賴，降低了延遲等待時(shí)間。內(nèi)存計(jì)算下，Spark 比 MapReduce 快100倍。

2，易用性

不同于MapReduce僅支持Map和Reduce兩種編程算子，Spark提供了超過80種不同的Transformation和Action算子，如map,reduce,filter,groupByKey,sortByKey,foreach等，并且采用函數(shù)式編程風(fēng)格，實(shí)現(xiàn)相同的功能需要的代碼量極大縮小。

3，通用性

Spark提供了統(tǒng)一的解決方案。Spark可以用于批處理、交互式查詢（Spark SQL）、實(shí)時(shí)流處理（Spark Streaming）、機(jī)器學(xué)習(xí)（Spark MLlib）和圖計(jì)算（GraphX）。

這些不同類型的處理都可以在同一個(gè)應(yīng)用中無縫使用。這對(duì)于企業(yè)應(yīng)用來說，就可使用一個(gè)平臺(tái)來進(jìn)行不同的工程實(shí)現(xiàn)，減少了人力開發(fā)和平臺(tái)部署成本。

4，兼容性

Spark能夠跟很多開源工程兼容使用。如Spark可以使用Hadoop的YARN和Apache Mesos作為它的資源管理和調(diào)度器，并且Spark可以讀取多種數(shù)據(jù)源，如HDFS、HBase、MySQL等。

二，Spark基本概念

RDD：是彈性分布式數(shù)據(jù)集（Resilient Distributed Dataset）的簡(jiǎn)稱，是分布式內(nèi)存的一個(gè)抽象概念，提供了一種高度受限的共享內(nèi)存模型。

DAG：是Directed Acyclic Graph（有向無環(huán)圖）的簡(jiǎn)稱，反映RDD之間的依賴關(guān)系。

Driver Program：控制程序，負(fù)責(zé)為Application構(gòu)建DAG圖。

Cluster Manager：集群資源管理中心，負(fù)責(zé)分配計(jì)算資源。

Worker Node：工作節(jié)點(diǎn)，負(fù)責(zé)完成具體計(jì)算。

Executor：是運(yùn)行在工作節(jié)點(diǎn)（Worker Node）上的一個(gè)進(jìn)程，負(fù)責(zé)運(yùn)行Task，并為應(yīng)用程序存儲(chǔ)數(shù)據(jù)。

Application：用戶編寫的Spark應(yīng)用程序，一個(gè)Application包含多個(gè)Job。

Job：作業(yè)，一個(gè)Job包含多個(gè)RDD及作用于相應(yīng)RDD上的各種操作。

Stage：階段，是作業(yè)的基本調(diào)度單位，一個(gè)作業(yè)會(huì)分為多組任務(wù)，每組任務(wù)被稱為“階段”。

Task：任務(wù)，運(yùn)行在Executor上的工作單元，是Executor中的一個(gè)線程。

總結(jié)：Application由多個(gè)Job組成，Job由多個(gè)Stage組成，Stage由多個(gè)Task組成。Stage是作業(yè)調(diào)度的基本單位。

三，Spark架構(gòu)設(shè)計(jì)

Spark集群由Driver, Cluster Manager（Standalone,Yarn 或 Mesos），以及Worker Node組成。對(duì)于每個(gè)Spark應(yīng)用程序，Worker Node上存在一個(gè)Executor進(jìn)程，Executor進(jìn)程中包括多個(gè)Task線程。

對(duì)于pyspark,為了不破壞Spark已有的運(yùn)行時(shí)架構(gòu)，Spark在外圍包裝一層Python API。在Driver端，借助Py4j實(shí)現(xiàn)Python和Java的交互，進(jìn)而實(shí)現(xiàn)通過Python編寫Spark應(yīng)用程序。在Executor端，則不需要借助Py4j，因?yàn)镋xecutor端運(yùn)行的Task邏輯是由Driver發(fā)過來的，那是序列化后的字節(jié)碼。

四，Spark運(yùn)行流程

1，Application首先被Driver構(gòu)建DAG圖并分解成Stage。

2，然后Driver向Cluster Manager申請(qǐng)資源。

3，Cluster Manager向某些Work Node發(fā)送征召信號(hào)。

4，被征召的Work Node啟動(dòng)Executor進(jìn)程響應(yīng)征召，并向Driver申請(qǐng)任務(wù)。

5，Driver分配Task給Work Node。

6，Executor以Stage為單位執(zhí)行Task，期間Driver進(jìn)行監(jiān)控。

7，Driver收到Executor任務(wù)完成的信號(hào)后向Cluster Manager發(fā)送注銷信號(hào)。

8，Cluster Manager向Work Node發(fā)送釋放資源信號(hào)。

9，Work Node對(duì)應(yīng)Executor停止運(yùn)行。

五，Spark部署模式

Local：本地運(yùn)行模式，非分布式。

Standalone：使用Spark自帶集群管理器，部署后只能運(yùn)行Spark任務(wù)。

Yarn：Haoop集群管理器，部署后可以同時(shí)運(yùn)行MapReduce，Spark，Storm，Hbase等各種任務(wù)。

Mesos：與Yarn最大的不同是Mesos 的資源分配是二次的，Mesos負(fù)責(zé)分配一次，計(jì)算框架可以選擇接受或者拒絕。

六，RDD數(shù)據(jù)結(jié)構(gòu)

RDD全稱Resilient Distributed Dataset，彈性分布式數(shù)據(jù)集，它是記錄的只讀分區(qū)集合，是Spark的基本數(shù)據(jù)結(jié)構(gòu)。

RDD代表一個(gè)不可變、可分區(qū)、里面的元素可并行計(jì)算的集合。

一般有兩種方式創(chuàng)建RDD，第一種是讀取文件中的數(shù)據(jù)生成RDD，第二種則是通過將內(nèi)存中的對(duì)象并行化得到RDD。

#通過讀取文件生成RDD
rdd = sc.textFile("hdfs://hans/data_warehouse/test/data")


#通過將內(nèi)存中的對(duì)象并行化得到RDD
arr = [1,2,3,4,5]
rdd = sc.parallelize(arr)

創(chuàng)建RDD之后，可以使用各種操作對(duì)RDD進(jìn)行編程。

RDD的操作有兩種類型，即Transformation操作和Action操作。轉(zhuǎn)換操作是從已經(jīng)存在的RDD創(chuàng)建一個(gè)新的RDD，而行動(dòng)操作是在RDD上進(jìn)行計(jì)算后返回結(jié)果到 Driver。

Transformation操作都具有 Lazy 特性，即 Spark 不會(huì)立刻進(jìn)行實(shí)際的計(jì)算，只會(huì)記錄執(zhí)行的軌跡，只有觸發(fā)Action操作的時(shí)候，它才會(huì)根據(jù) DAG 圖真正執(zhí)行。

操作確定了RDD之間的依賴關(guān)系。

RDD之間的依賴關(guān)系有兩種類型，即窄依賴和寬依賴。窄依賴時(shí)，父RDD的分區(qū)和子RDD的分區(qū)的關(guān)系是一對(duì)一或者多對(duì)一的關(guān)系。而寬依賴時(shí)，父RDD的分區(qū)和自RDD的分區(qū)是一對(duì)多或者多對(duì)多的關(guān)系。

寬依賴關(guān)系相關(guān)的操作一般具有shuffle過程，即通過一個(gè)Patitioner函數(shù)將父RDD中每個(gè)分區(qū)上key不同的記錄分發(fā)到不同的子RDD分區(qū)。

依賴關(guān)系確定了DAG切分成Stage的方式。

切割規(guī)則：從后往前，遇到寬依賴就切割Stage。

RDD之間的依賴關(guān)系形成一個(gè)DAG有向無環(huán)圖，DAG會(huì)提交給DAGScheduler，DAGScheduler會(huì)把DAG劃分成相互依賴的多個(gè)stage，劃分stage的依據(jù)就是RDD之間的寬窄依賴。遇到寬依賴就劃分stage,每個(gè)stage包含一個(gè)或多個(gè)task任務(wù)。然后將這些task以taskSet的形式提交給TaskScheduler運(yùn)行。

七，WordCount范例

import findspark

#指定spark_home為剛才的解壓路徑,指定python路徑
spark_home = "/Users/liangyun/ProgramFiles/spark-3.0.1-bin-hadoop3.2"
python_path = "/Users/liangyun/anaconda3/bin/python"
findspark.init(spark_home,python_path)

import pyspark 
from pyspark import SparkContext, SparkConf
conf = SparkConf().setAppName("test").setMaster("local[4]")
sc = SparkContext(conf=conf)

只需要5行代碼就可以完成WordCount詞頻統(tǒng)計(jì)。

rdd_line = sc.textFile("./data/hello.txt")
rdd_word = rdd_line.flatMap(lambda x:x.split(" "))
rdd_one = rdd_word.map(lambda t:(t,1))
rdd_count = rdd_one.reduceByKey(lambda x,y:x+y)
rdd_count.collect() 

[('world', 1),
 ('love', 3),
 ('jupyter', 1),
 ('pandas', 1),
 ('hello', 2),
 ('spark', 4),
 ('sql', 1)]