本文主要翻譯至鏈接且不局限于該文內(nèi)容狞甚，也加入了筆者實(shí)踐內(nèi)容锁摔，翻譯水平有限，歡迎指正哼审，轉(zhuǎn)載請(qǐng)注明出處谐腰。
概述
每個(gè)Spark應(yīng)用程序都包含了一個(gè)驅(qū)動(dòng)程序孕豹，用于執(zhí)行用戶編寫的main函數(shù)，以及在集群上執(zhí)行各種并行操作怔蚌。Spark抽象了一個(gè)稱為RDD（resilient distributed dataset，彈性分布式數(shù)據(jù)集）的數(shù)據(jù)集旁赊，它是一個(gè)數(shù)據(jù)元素的集合桦踊，這些元素被分散在整個(gè)集群的各個(gè)節(jié)點(diǎn)上，可以進(jìn)行并行操作终畅。RDDs可以從Hadoop的文件系統(tǒng)中創(chuàng)建（HDFS籍胯，或其他Hadoop支持的文件系統(tǒng)），或者從驅(qū)動(dòng)程序中已存在的Scala集合創(chuàng)建离福，然后進(jìn)行轉(zhuǎn)換杖狼。用戶可以在Spark中設(shè)置將RDD駐留在內(nèi)存，以便在并行操作中可以高效重用妖爷。此外蝶涩，RDDs也支持從節(jié)點(diǎn)故障中自動(dòng)恢復(fù)。
Spark還抽象了一個(gè)在并行操作中使用的稱為“shared variables”（共享變量）的概念絮识。默認(rèn)情況下绿聘，當(dāng)Spark在集群的不同節(jié)點(diǎn)中執(zhí)行一系列并發(fā)作業(yè)時(shí)（它們執(zhí)行一個(gè)相同的函數(shù)），它將會(huì)把函數(shù)中的變量拷貝到每一個(gè)作業(yè)上次舌。這些變量有時(shí)候需要在作業(yè)之間共享熄攘，有時(shí)候也會(huì)在作業(yè)和驅(qū)動(dòng)程序之間共享。Spark支持兩種類型的變量：1）廣播變量（broadcast variables）彼念，可以當(dāng)做是所有節(jié)點(diǎn)內(nèi)存中的一個(gè)緩存值挪圾；2）累加器（Accumulators），這類變量只允許“加減”逐沙，如同計(jì)數(shù)器哲思。
本篇指引描述了Spark支持的三種語(yǔ)言編程方式，包括Scala吩案、Java和Pyhon也殖，為了方便閱讀，筆者將其拆成多篇文章务热，其中Python版本暫不翻譯忆嗜。本篇為Scala版，由于文章較長(zhǎng)崎岂，上下兩篇捆毫。
如果你要跟著編程指引進(jìn)行學(xué)習(xí)，建議你使用命令行交互方式啟動(dòng)Spark（對(duì)于Scala冲甘，使用Spark-shell命令绩卤，對(duì)于Python途样，使用pyspark命令）,或者在IDE編寫代碼進(jìn)行學(xué)習(xí)。
引入Spark
Spark2.1.0默認(rèn)與Scala2.11一起發(fā)布濒憋，當(dāng)然也可以和其他版本的Scala一起工作何暇。為了保證程序能夠正常執(zhí)行，你最好使用一個(gè)Scala兼容版本（例如：2.11.X凛驮，應(yīng)該謹(jǐn)慎選擇jdk和scala的對(duì)應(yīng)版本裆站，例如2.12版本的scala只支持java8，較老的scala版本兼容jdk6+黔夭，而java9還不支持使用scala宏胯，參考scala）。
筆者安裝的Spark版本為2.1.0本姥，使用的scala版本為2.11.7肩袍，jdk為7.0。
開始編寫Spark應(yīng)用程序前婚惫，你需要先創(chuàng)建一個(gè)工程氛赐，然后引入相關(guān)的依賴，可以使用sbt構(gòu)建你的工程先舷，筆者在IntelliJ中安裝了scala插件鹰祸，然后基于sbt新建了一個(gè)工程，如圖：
之后密浑，在build.sbt中編寫項(xiàng)目依賴以引入spark相關(guān)的jar包：
name := "scala-demo"
version := "1.0"
scalaVersion := "2.11.7"
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"
在這個(gè)文件中蛙婴，我們指定了scala的版本以及依賴的庫(kù)，sbt會(huì)自動(dòng)解析這個(gè)依賴尔破，并從遠(yuǎn)程中央倉(cāng)庫(kù)中下載相應(yīng)的jar包街图。
關(guān)于如何在IntelliJ中使用Spark，可以參考這篇博客懒构。
注意餐济，在IntelliJ中編寫的Spark程序不能夠直接提交到Spark集群上執(zhí)行，但可以簡(jiǎn)單的在“l(fā)ocal”上調(diào)試胆剧，如果需要提交到Spark集群上絮姆，應(yīng)該先構(gòu)建成jar包，然后使用spark-submit提交秩霍。
初始化Spark
Spark應(yīng)用程序的第一步是創(chuàng)建一個(gè)SparkContext對(duì)象篙悯，它將用來(lái)連接Spark集群。SparkContext的創(chuàng)建需要一個(gè)SparkConf對(duì)象作為參數(shù)铃绒，這個(gè)對(duì)象包含了應(yīng)用程序的詳細(xì)信息鸽照。
一個(gè)JVM只能使用一個(gè)SparkContext，在創(chuàng)建新的SparkContext之前颠悬，你需要調(diào)用stop()方法來(lái)停止活躍著的SparkContext矮燎。下面是一個(gè)創(chuàng)建實(shí)例：
val conf = new SparkConf().setAppName(appName).setMaster(master)
new SparkContext(conf)
在這個(gè)例子中定血，appName指應(yīng)用程序的名稱，它將展示在集群的webUI上诞外。master指的是Spark集群的地址（可以是三種運(yùn)行模式之一澜沟，Spark獨(dú)立模式，Mesos模式和YARN模式）峡谊，或者直接用“l(fā)ocal”表示本地的Spark茫虽。實(shí)踐中，你可能不會(huì)直接把master硬編碼在代碼中靖苇，而是使用spark-submit命令時(shí)實(shí)時(shí)指定席噩，但是如果是本地測(cè)試和單元測(cè)試班缰，傳遞“l(fā)ocal”以在本地運(yùn)行可以提高效率贤壁。
使用Shell命令行
在Spark的shell環(huán)境中，已經(jīng)默認(rèn)為你創(chuàng)建了一個(gè)SparkContext埠忘，變量名為“sc”脾拆，你不必重新創(chuàng)建，事實(shí)上重新創(chuàng)建的SparkContext也無(wú)法工作莹妒。你可以在執(zhí)行命令時(shí)使用--master選項(xiàng)配置master的地址名船，例如：
./bin/spark-shell --master local[4]
或者，使用--jars選項(xiàng)將jars文件添加的classpath中旨怠，多個(gè)jars文件之間使用英文逗號(hào)分隔渠驼，例如：
./bin/spark-shell --master local[4] --jars code.jar,code2.jar
甚至你可以使用--packages選項(xiàng)直接添加依賴庫(kù)的Maven坐標(biāo)，多個(gè)依賴庫(kù)之間使用逗號(hào)分隔鉴腻，如果有附加的代碼倉(cāng)庫(kù)迷扇，可以使用--repositories引入。例如：
./bin/spark-shell --master local[4] --packages "org.example:example:0.1"
關(guān)于spark-shell腳本爽哎，可以在筆者翻譯的《Spark指南》三蜓席、 提交應(yīng)用程序這篇文章中找到更多的講解。
彈性分布式數(shù)據(jù)集（RDDs）
Spark圍繞著一個(gè)稱為彈性分布式數(shù)據(jù)集（RDD）的虛擬概念來(lái)操作數(shù)據(jù)课锌，這是一個(gè)可以并行操作的可容錯(cuò)的元素集合厨内。 有兩種方法來(lái)創(chuàng)建RDD：并行化驅(qū)動(dòng)程序中的現(xiàn)有集合，或引用外部存儲(chǔ)系統(tǒng)中的數(shù)據(jù)集渺贤，例如共享文件系統(tǒng)雏胃，HDFS，HBase或提供Hadoop InputFormat的任何數(shù)據(jù)源志鞍。
并行集合
可以通過(guò)以內(nèi)存一個(gè)已存在的集合為數(shù)據(jù)集（Scala Seq）丑掺，然后調(diào)用SparkContext的parallelize方法創(chuàng)建一個(gè)并行集合。該集合的數(shù)據(jù)將被拷貝到集群中述雾，成為一個(gè)分布式數(shù)據(jù)集街州。例如兼丰，
val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)
一旦創(chuàng)建成功，這個(gè)分布式數(shù)據(jù)集就可以被并行操作唆缴。例如鳍征，我們可以調(diào)用distData.reduce((a,b) => a+b)來(lái)計(jì)算該數(shù)據(jù)集的累加和。
并行集合的一個(gè)重要參數(shù)是數(shù)據(jù)集在集群中被拆分成的分區(qū)數(shù)面徽，Spark將會(huì)為每個(gè)分區(qū)單獨(dú)創(chuàng)建一個(gè)task艳丛，通常，可以為集群中的每個(gè)CPU分配2~4個(gè)分區(qū)趟紊，如果不加指定氮双，Spark會(huì)根據(jù)你集群中的配置自動(dòng)設(shè)置分區(qū)數(shù)。如果你想手動(dòng)指定霎匈，只要在parallelize函數(shù)中傳遞參數(shù)戴差，例如sc.parallelize(data, 10)。注意铛嘱，代碼中的一些地方使用術(shù)語(yǔ)slice（分區(qū)的同義詞）來(lái)保持向后兼容性暖释。
外部數(shù)據(jù)集
Spark可以從任意Hadoop支持的數(shù)據(jù)源中創(chuàng)建分布式數(shù)據(jù)集，包括本地文件系統(tǒng)墨吓，HDFS球匕，Cassandra，HBase帖烘，Amazon S3等等亮曹。Spark支持文本文件，SequenceFiles 和其他任意的Hadoop InputFormat秘症。
文本文件的RDDs可以通過(guò)SparkContext的textFile方法進(jìn)行創(chuàng)建照卦，這個(gè)方法使用文件的URI作為參數(shù)（例如，本地文件路徑历极，hdfs://窄瘟，s3n:// 等URI），文件被創(chuàng)建后趟卸，將被構(gòu)造成一個(gè)集合蹄葱，每一行作為一個(gè)元素。下面是一個(gè)調(diào)用例子：
scala> val distFile = sc.textFile("data.txt")
distFile: org.apache.spark.rdd.RDD[String] = data.txt MapPartitionsRDD[10] at textFile at <console>:26
一旦創(chuàng)建成功锄列，就可以在distFile上使用各種數(shù)據(jù)集的操作图云，例如，可以通過(guò)下面的代碼統(tǒng)計(jì)文件的字符數(shù)：
scala> distFile.map(s => s.length).reduce((a, b) => a + b)
在Spark中讀取文件應(yīng)該注意如下幾個(gè)事項(xiàng)：
- 如果使用本地文件系統(tǒng)的路徑邻邮，并提交到集群竣况，該文件必須同時(shí)在工作節(jié)點(diǎn)中存在，可以拷貝文件到各個(gè)工作節(jié)點(diǎn)筒严，或者使用基于網(wǎng)絡(luò)的共享文件系統(tǒng)丹泉。
- 所有基于文件的輸入方法情萤，包括textFile方法，都支持以目錄摹恨、壓縮文件和通配符指定的文件筋岛，例如可以創(chuàng)建textFile("/my/directory"), textFile("/my/directory/*.txt"), and textFile("/my/directory/*.gz")。
- textFile方法支持可選的參數(shù)晒哄，用于控制文件的分區(qū)數(shù)睁宰。默認(rèn)情況下，Spark給每個(gè)文件塊創(chuàng)建一個(gè)分區(qū)（HDFS中寝凌，文件塊的默認(rèn)大小為128M）柒傻，你可以傳遞一個(gè)其他的數(shù)值來(lái)指定分區(qū)數(shù)的大小。注意较木，自己指定時(shí)红符，分區(qū)數(shù)不可以小于文件塊數(shù)。
除了文本文件劫映，Spark的Scala API也支持其他的一些數(shù)據(jù)格式：
- SparkContext.wholeTextFiles 允許你讀取一個(gè)包含多個(gè)小文件的目錄违孝，然后以(filename, content) pairs返回刹前。這個(gè)方法和textFile不一致泳赋，textFile返回的格式中，每一行是一條記錄喇喉。
- 對(duì)于其他Hadoop InputFormats的文件祖今，你可以使用SparkContext.hadoopRDD方法，該方法接受任意的JobConf和輸入格式類拣技，鍵類和值類千诬。使用時(shí)，與使用輸入源的Hadoop作業(yè)相同的方式進(jìn)行設(shè)置膏斤。 你還可以使用SparkContext.newAPIHadoopRDD徐绑，用于創(chuàng)建基于“新”MapReduce API（org.apache.hadoop.mapreduce）的InputFormats。
- RDD.saveAsObjectFile和SparkContext.objectFile方法支持以包含序列化Java對(duì)象的簡(jiǎn)單格式保存RDD莫辨。 它提供了一種保存RDD的簡(jiǎn)單方法傲茄。
RDD操作
RDDs支持兩種操作：1）transformations，這類操作對(duì)一個(gè)已存在的RDD進(jìn)行轉(zhuǎn)換操作沮榜，然后產(chǎn)生新的RDD盘榨；2）actions，這類操作對(duì)一個(gè)數(shù)據(jù)集進(jìn)行計(jì)算蟆融，然后返回一個(gè)值給驅(qū)動(dòng)程序草巡。例如，map函數(shù)就是一個(gè)transformation方法型酥，它以把一個(gè)數(shù)據(jù)集的每個(gè)元素傳遞個(gè)一個(gè)開發(fā)者定義的函數(shù)山憨，然后返回一個(gè)經(jīng)過(guò)處理的新的RDD查乒。而reduce函數(shù)就是一個(gè)action，它聚集一個(gè)RDD里的所有元素郁竟，對(duì)其進(jìn)行某個(gè)操作（同樣由開發(fā)者定義）侣颂，然后給驅(qū)動(dòng)程序返回最終的結(jié)果。
Spark中的所有transformations都是延時(shí)的枪孩，即它們不會(huì)立刻計(jì)算出結(jié)果憔晒， 相反，他們只記住應(yīng)用于一些基礎(chǔ)數(shù)據(jù)集（例如文件）的transformations蔑舞。 僅當(dāng)一個(gè)action需要將結(jié)果返回到驅(qū)動(dòng)程序時(shí)才會(huì)計(jì)算對(duì)應(yīng)的transformation拒担。 這種設(shè)計(jì)使Spark能夠更高效地運(yùn)行。
默認(rèn)情況下攻询，每一個(gè)transformed RDD在運(yùn)行一個(gè)action時(shí)都會(huì)重新進(jìn)行計(jì)算从撼，但是，你可以使用Spark提供的persist或cache方法將一個(gè)RDD保留在集群機(jī)器的內(nèi)存中钧栖，這樣下次就可以更快的訪問(wèn)它們低零。Spark還支持在磁盤上持久化存儲(chǔ)RDD，或者在多個(gè)節(jié)點(diǎn)上復(fù)制RDD拯杠。
RDD基礎(chǔ)
為了說(shuō)明RDD入門掏婶，參考如下簡(jiǎn)單的程序：
val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)
第一行以外部文件定義了一個(gè)基本的RDD。此時(shí)lines僅僅是指向文件的一個(gè)指針潭陪，還未導(dǎo)入內(nèi)存雄妥。第二行對(duì)這個(gè)數(shù)據(jù)集進(jìn)行了map transformation，提交的函數(shù)用于計(jì)算每一行的字符長(zhǎng)度依溯，此時(shí)老厌，由于Spark的lazy特性，還未立刻進(jìn)行計(jì)算黎炉。第三行枝秤，執(zhí)行了一個(gè)reduce action，函數(shù)是用來(lái)累加多行的字符長(zhǎng)度和慷嗜，此時(shí)淀弹，Spark將計(jì)算任務(wù)拆成多個(gè)task以在多個(gè)獨(dú)立的機(jī)器上執(zhí)行，在每臺(tái)機(jī)器上都只對(duì)應(yīng)的對(duì)拷貝到本地的數(shù)據(jù)子集進(jìn)行map和reduce操作洪添，然后將結(jié)果返回給驅(qū)動(dòng)程序垦页。
如果我們希望下次繼續(xù)使用lineLengths，我們需要在reduce方法前添加如下代碼：
lineLengths.persist()
使用本方法后干奢，lineLengths的結(jié)果將在第一次計(jì)算后保留在節(jié)點(diǎn)的內(nèi)存中痊焊。
傳遞Functions給Spark
Spark上的計(jì)算強(qiáng)烈依賴于提交給它的function，編寫代碼時(shí)，推薦一下兩種方式：
- 匿名函數(shù)語(yǔ)法薄啥，推薦在較短的代碼中使用辕羽。
- 以一個(gè)全局單例的靜態(tài)方法提交，例如垄惧，以你可以像下面這樣定義object MyFunctions刁愿，然后將MyFunctions.func1作為參數(shù)傳遞：
object MyFunctions {
def func1(s: String): String = { ... }
}
myRdd.map(MyFunctions.func1)
注意，雖然Spark允許把一個(gè)類示例的方法引用作為Function（與單例object方式相反）到逊，但是這種方式需要把包含該類的對(duì)象和方法發(fā)送到集群中铣口。例如下面這種方式的調(diào)用：
class MyClass {
def func1(s: String): String = { ... }
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
}
如果我們創(chuàng)建了一個(gè)MyClass實(shí)例，然后調(diào)用該實(shí)例的doStuff方法觉壶，由于map操作引用了該實(shí)例的func1方法脑题，因此整個(gè)實(shí)例都需要被發(fā)送到集群中（維持中間狀態(tài)）。執(zhí)行 rdd.map(x => this.func1(x)) 也是類似的效果铜靶。
相似的叔遂，訪問(wèn)外部對(duì)象的字段，也會(huì)間接引用整個(gè)對(duì)象争剿，例如：
class MyClass {
val field = "Hello"
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
}
上面這個(gè)Function相當(dāng)于 rdd.map(x => this.field + x) 已艰。為了避免這個(gè)問(wèn)題，最簡(jiǎn)單的方式是把字段拷貝成一個(gè)局部變量蚕苇，而不是引用外部的字段哩掺。例如：
def doStuff(rdd: RDD[String]): RDD[String] = {
val field_ = this.field
rdd.map(x => field_ + x)
}
理解閉包
Spark的學(xué)習(xí)難點(diǎn)是當(dāng)代碼在跨集群執(zhí)行時(shí)，理解變量和方法的作用域和生命周期捆蜀。RDD對(duì)一些超出作用域的變量進(jìn)行的操作疮丛，會(huì)給開發(fā)者造成一些困惑幔嫂，下面通過(guò)一個(gè)實(shí)際的例子來(lái)仔細(xì)闡述這些細(xì)節(jié)辆它。
var counter = 0
var rdd = sc.parallelize(data)
// Wrong: 不要這樣做
rdd.foreach(x => counter += x)
println("Counter value: " + counter)
上面的這個(gè)例子如果在不同的JVM中執(zhí)行，將會(huì)產(chǎn)生不同的結(jié)果履恩，例如它在local模式和集群模式下的運(yùn)行結(jié)果將會(huì)不同锰茉，我們進(jìn)一步解釋這個(gè)原因。
本地模式&集群模式
Spark在執(zhí)行作業(yè)時(shí)切心，會(huì)把對(duì)RDD的操作拆分成多個(gè)子任務(wù)飒筑，每個(gè)子任務(wù)都會(huì)有一個(gè)執(zhí)行器。在執(zhí)行器執(zhí)行代碼之前绽昏，Spark會(huì)計(jì)算該子任務(wù)的閉包协屡，所謂的閉包，就是在操作RDD時(shí)（本例子中的操作即foreach()）所需要的變量和方法必須對(duì)執(zhí)行器可見全谤。該閉包被序列化后發(fā)送到每一個(gè)執(zhí)行器上肤晓。
閉包中的變量在發(fā)送給執(zhí)行器時(shí)會(huì)被拷貝，因此當(dāng)變量counter在foreach函數(shù)中被引用時(shí)，已經(jīng)不是驅(qū)動(dòng)器節(jié)點(diǎn)上的counter變量了补憾。雖然驅(qū)動(dòng)器節(jié)點(diǎn)的內(nèi)存中也有一個(gè)counter變量漫萄，但這個(gè)變量對(duì)其他的執(zhí)行器是不可見的，執(zhí)行器只能看見序列化后的閉包拷貝的副本盈匾。于是腾务，這段代碼執(zhí)行的最后結(jié)果仍然是0，因?yàn)閳?zhí)行器節(jié)點(diǎn)執(zhí)行的操作都是在它們序列化后的couter變量上削饵。
而在local模式下岩瘦，foreach函數(shù)的執(zhí)行器與驅(qū)動(dòng)程序使用的是同一個(gè)JVM，因此它們將會(huì)引用同一個(gè)counter變量窿撬，更新操作也能夠正常執(zhí)行担钮。
如果想要正常執(zhí)行，建議開發(fā)者使用Accumulator
（在《Spark指南》四尤仍、編程指引-Scala篇(下)一文中將詳細(xì)描述）箫津。Spark中的Accumulators累加器提供了一種在集群中更新變量的安全方式寝杖。
通常回季，閉包中構(gòu)造的循環(huán)或者本地方法，不應(yīng)該被用于更改某些全局的變量彬犯。Spark無(wú)法保證閉包之外對(duì)這些引用對(duì)象的更改行為赡模，有一些代碼可以在local模式下正常工作田炭，但這僅僅是巧合，一旦到了分布式模式下漓柑，可能這些代碼就無(wú)法工作了教硫。如果需要一些全局的聚合，建議使用Accumulator辆布。
打印RDD的元素
另一個(gè)常見的習(xí)慣是使用rdd.foreach(println) 或者 rdd.map(println)來(lái)嘗試打印一個(gè)RDD的輸出瞬矩。在單機(jī)模式下，所有的元素都可以正常的被打印出來(lái)锋玲，但是在集群模式中景用，所有的元素都被打印到執(zhí)行器的控制臺(tái)中，而不是在驅(qū)動(dòng)程序的機(jī)器上惭蹂。如果想達(dá)到這個(gè)目的伞插，你可以先使用collect()方法將所有元素聚集到驅(qū)動(dòng)機(jī)器上，然后執(zhí)行print操作盾碗，例如：rdd.collect().foreach(println)媚污。然而，這可能會(huì)使驅(qū)動(dòng)機(jī)器產(chǎn)生OOM（內(nèi)存溢出）錯(cuò)誤廷雅，因?yàn)檫@一操作會(huì)將整個(gè)RDD集合都提取到同一臺(tái)機(jī)器上耗美。如果你只是想打印少數(shù)元素氢伟，建議你使用諸如rdd.take(100).foreach(println)的代碼。
備注：鑒于篇幅太長(zhǎng)幽歼，此篇文章拆成兩篇來(lái)翻譯朵锣，下一篇請(qǐng)參考筆者文集中的《Spark指南》四、編程指引-Scala篇(下)甸私。
