spark源碼解析

A. 《Spark技術內幕深入解析Spark內核架構設計與實現原理》epub下載在線閱讀，求百度網盤雲資源

《Spark技術內幕》（張安站）電子書網盤下載免費在線閱讀

資源鏈接：

鏈接：https://pan..com/s/12ee-1OhjbK6DQgTdUzwK9Q

書名：Spark技術內幕

作者：張安站

豆瓣評分：7.6

出版社：機械工業出版社

出版年份：2015-9-1

頁數：201

內容簡介：

Spark是不斷壯大的大數據分析解決方案家族中備受關注的新增成員。它不僅為分布式數據集的處理提供一個有效框架，而且以高效的方式處理分布式數據集。它支持實時處理、流處理和批處理，提供了AllinOne的統一解決方案，使得Spark極具競爭力。

本書以源碼為基礎，深入分析Spark內核的設計理念和架構實現，系統講解各個核心模塊的實現，為性能調優、二次開發和系統運維提供理論支持；本文最後以項目實戰的方式，系統講解生產環境下Spark應用的開發、部署和性能調優。

作者簡介：

張安站，2008年本科畢業於天津大學，2011年研究生畢業於南開大學，畢業後加入EMC中國卓越研發集團，從事商業存儲系統控制模塊的研發。2014年7月加入網路上海研發中心，從事分布式系統架構相關工作，當前負責NLP Cloud（為公司內用戶提供NLP核心演算法的雲平台）和Sofa Cloud（為策略服務化提供一站式解決方案的雲平台）兩個計算方向。目前專注分布式資源管理和大數據處理平台等相關技術。

B. spark mllib演算法介面源碼在什麼地方查看

1.1LDA實例實例步驟：1）載入數據返回的數據格式為：documents:RDD[(Long,Vector)]，其中：Long為文章ID，Vector為文章分詞後的詞向量；用戶可以讀取指定目錄下的數據，通過分詞以及數據格式的轉換，轉換成RDD[(Long,Vector)]即可。2）建立模型模型參數設置說明：k:主題數，或者聚類中心數DocConcentration：文章分布的超參數(Dirichlet分布的參數)，必需>1.0TopicConcentration：主題分布的超參數(Dirichlet分布的參數)，必需>1.0MaxIterations：迭代次數setSeed：隨機種子CheckpointInterval：迭代計算時檢查點的間隔Optimizer：優化計算方法，目前支持"em","online"3）結果輸出topicsMatrix以及topics(word,topic))輸出。實例代碼如下：[java]viewplainimportorg.apache.log4j.{Level,Logger}importorg.apache.spark.{SparkConf,SparkContext}importorg.apache.spark.mllib.clustering.LDAimportorg.apache.spark.mllib.linalg.Vectorsobjectlda{defmain(args:Array[String]){//0構建Spark對象valconf=newSparkConf().setAppName("lda")valsc=newSparkContext(conf)Logger.getRootLogger.setLevel(Level.WARN)//1載入數據，返回的數據格式為：documents:RDD[(Long,Vector)]//其中：Long為文章ID，Vector為文章分詞後的詞向量//可以讀取指定目錄下的數據，通過分詞以及數據格式的轉換，轉換成RDD[(Long,Vector)]即可valdata=sc.textFile("data/mllib/sample_lda_data.txt")valparsedData=data.map(s=>Vectors.dense(s.trim.split('').map(_.toDouble)))//=parsedData.zipWithIndex.map(_.swap).cache()//2建立模型，設置訓練參數，訓練模型/***k:主題數，或者聚類中心數*DocConcentration：文章分布的超參數(Dirichlet分布的參數)，必需>1.0*TopicConcentration：主題分布的超參數(Dirichlet分布的參數)，必需>1.0*MaxIterations：迭代次數*setSeed：隨機種子*CheckpointInterval：迭代計算時檢查點的間隔*Optimizer：優化計算方法，目前支持"em","online"*/valldaModel=newLDA().setK(3).setDocConcentration(5).setTopicConcentration(5).setMaxIterations(20).setSeed(0L).setCheckpointInterval(10).setOptimizer("em").run(corpus)//3模型輸出，模型參數輸出，結果輸出//Outputtopics.Eachisadistributionoverwords(matchingwordcountvectors)println("Learnedtopics(asdistributionsovervocabof"+ldaModel.vocabSize+"words):")valtopics=ldaModel.topicsMatrixfor(topic<-Range(0,3)){print("Topic"+topic+":")for(word<-Range(0,ldaModel.vocabSize)){print(""+topics(word,topic));}println()}}}

C. 如何構建第一個Spark項目代碼

操作系統
Window7/Mac
IDE
IntelliJ IDEA Community Edition 14.1.6
下載地址
JDK 1.8.0_65
下載地址
Scala 2.11.7
下載地址
其它環境
Spark:1.4.1
下載地址
Hadoop Yarn:Hadoop 2.5.0-cdh5.3.2
IDE項目創建
新建一個項目
New Project

使用Maven模型創建一個Scala項目

填寫自己的GroupId、ArtifactId,Version不需要修改，Maven會根據GroupId生成相應的目錄結構，GroupId的取值一般為a.b.c 結構,ArtifactId為項目名稱。之後點擊next，填寫完項目名稱和目錄，點擊finish就可以讓maven幫你創建Scala項目

項目創建完成後，目錄結構如下

4.為項目添加JDK以及Scala SDK
點擊File->Project Structure，在SDKS和Global Libraries中為項目配置環境。

至此整個項目結構、項目環境都搭建好了
編寫主函數
主函數的編寫在 projectName/src/main/scala/…/下完成，如果按照上述步驟完成代碼搭建，將在目錄最後發現
MyRouteBuild
MyRouteMain

這兩個文件為模塊文件，刪除MyRouteBuild,重命名MyRouteMain為DirectKafkaWordCount。這里，我使用Spark Streaming官方提供的一個代碼為實例代碼，代碼如下
package org.apache.spark.examples.streaming

import kafka.serializer.StringDecoder

import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.SparkConf

object DirectKafkaWordCount {
def main(args: Array[String]) {
if (args.length < 2) {
System.err.println("...")
System.exit(1)
}
//StreamingExamples.setStreamingLogLevels()

val Array(brokers, topics) = args

val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount")
val ssc = new StreamingContext(sparkConf, Seconds(2))

// Create direct kafka stream with brokers and topics
val topicsSet = topics.split(",").toSet
val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
ssc, kafkaParams, topicsSet)

// Get the lines, split them into words, count the words and print
val lines = messages.map(_._2)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1L)).receByKey(_ + _)
wordCounts.print()

// Start the computation
ssc.start()
ssc.awaitTermination()
}
}

將代碼最上面的package org.apache.spark.examples.streaming,替換為DirectKafkaWordCount里的package部分即可。並覆蓋DirectKafkaWordCount文件。
至此Spark處理代碼已經編寫完成。
修改pom.xml,為項目打包做准備
pom.xml中編寫了整個項目的依賴關系，這個項目中我們需要導入一些Spark Streaming相關的包。
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.10</artifactId>
<version>1.4.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka_2.10</artifactId>
<version>1.4.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
<version>1.4.1</version>
</dependency>

<!-- scala -->
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.10.4</version>
</dependency>

除此之外，如果需要把相關依賴打包到最終JAR包中，需要在pom.xml的bulid標簽中寫入以下配置:
<plugins>
<!-- Plugin to create a single jar that includes all dependencies -->
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.4</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>2.0.2</version>
<configuration>
<source>1.7</source>
<target>1.7</target>
</configuration>
</plugin>

<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<executions>
<execution>
<id>scala-compile-first</id>
<phase>process-resources</phase>
<goals>
<goal>add-source</goal>
<goal>compile</goal>
</goals>
</execution>
<execution>
<id>scala-test-compile</id>
<phase>process-test-resources</phase>
<goals>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>

pom.xml文件修改完成後，即可開始maven打包,操作如圖:

點擊右側彈出窗口的Execute Maven Goal,在command line中輸入clean package

Spark作業提交
在項目projectname/target目錄下即可找到兩個jar包，其中一個僅包含Scala代碼，另一個包含所有依賴的包。
將jar包導到Spark伺服器，運行Spark作業，運行操作如下
../bin/spark-submit –master yarn-client –jars ../lib/kafka_2.10-0.8.2.1.jar –class huochen.spark.example.DirectKafkaWordCount sparkExample-1.0-SNAPSHOT-jar-with-dependencies.jar kafka-broker topic
利用spark-submit把任務提交到Yarn集群，即可看到運行結果。

D. 《深入理解SPARK核心思想與源碼分析》epub下載在線閱讀，求百度網盤雲資源

《深入理解SPARK》（耿嘉安）電子書網盤下載免費在線閱讀

資源鏈接：

鏈接：

書名：深入理解SPARK

作者：耿嘉安

豆瓣評分：7.2

出版社：機械工業出版社

出版年份：2016-1-1

頁數：469

內容簡介：

《深入理解SPARK：核心思想與源碼分析》結合大量圖和示例，對Spark的架構、部署模式和工作模塊的設計理念、實現源碼與使用技巧進行了深入的剖析與解讀。

《深入理解SPARK：核心思想與源碼分析》一書對Spark1.2.0版本的源代碼進行了全面而深入的分析，旨在為Spark的優化、定製和擴展提供原理性的指導。阿里巴巴集團專家鼎力推薦、阿里巴巴資深Java開發和大數據專家撰寫。

本書分為三篇：

准備篇（第1～2章），介紹了Spark的環境搭建、設計理念與基本架構，幫助讀者了解一些背景知識。

核心設計篇（第3～7章），著重講解SparkContext的初始化、存儲體系、任務提交與執行、計算引擎及部署模式的原理和源碼分析。通過這部分的內容，讀者可以通過源碼剖析更加深入理解Spark的核心設計與實現，以便在實際使用中能夠快速解決線上問題並對性能進行調優。

擴展篇（第8～11章），主要講解基於Spark核心的各種擴展及應用，包括SQL處理引擎、Hive處理、流式計算框架Spark Streaming、圖計算框架GraphX、機器學習庫MLlib等內容。通過閱讀這部分內容，讀者可以擴展實際項目中對Spark的應用場景，讓Spark煥發活力。

作者簡介：

耿嘉安，10年IT行業相關經驗。就職於阿里巴巴商家業務事業部，任資深Java工程師，專注於開源和大數據領域，目前與小夥伴們基於ODPS構建阿里的大數據商業解決方案——御膳房。在大量的工作實踐中，對J2EE、JVM、Tomcat、Spring、Hadoop、Spark、MySQL、Redis都有深入研究，尤其喜歡剖析開源項目的源碼實現。早期從事J2EE企業級應用開發，對Java相關技術有獨到見解。業余時間喜歡研究中國古代歷史，古詩詞，旅遊，足球等。

E. 怎樣成為Spark高手

1.熟練掌握Scala語言
Spark框架是採用Scala語言編寫的，精緻而優雅。要想成為Spark高手，你就必須閱讀Spark的源代碼，就必須掌握Scala;
雖然說現在的Spark可以採用多語言Java、python等進行應用程序開發，但是最快速的和支持最好的開發API依然並將永遠是Scala方式的API，所以你必須掌握Scala來編寫復雜的和高性能的Spark分布式程序;
尤其要熟練掌握Scala的trait、apply、函數式編程、泛型、逆變與協變等;
2.精通Spark平台API
掌握Spark中面向RDD的開發模式，掌握各種transformation和action函數的使用;
掌握Spark中的寬依賴和窄依賴以及lineage機制;
掌握RDD的計算流程，例如Stage的劃分、Spark應用程序提交給集群的基本過程和Worker節點基礎的工作原理等。
3.深入Spark內核
此階段主要是通過Spark框架的源碼研讀來深入Spark內核部分：
通過源碼掌握Spark的任務提交過程;
通過源碼掌握Spark集群的任務調度;
尤其要精通DAGScheler、TaskScheler和Worker節點內部的工作的每一步的細節;
4.掌握基於Spark的核心框架
Spark作為雲計算大數據時代的集大成者，在實時流處理、圖技術、機器學習、NoSQL查詢等方面具有顯著的優勢，我們使用Spark的時候大部分時間都是在使用其上的框架例如Shark、Spark Streaming等：
Spark Streaming是非常出色的實時流處理框架，要掌握其DStream、transformation和checkpoint等;
Spark的離線統計分析功能，Spark1.0.0版本在Shark的基礎上推出了SparkSQL，離線統計分析的功能的效率有顯著的提升，
對於Spark的機器學習和GraphX等要掌握其原理和用法;
5.做商業級別的Spark項目
通過一個完整的具有代表性的Spark項目來貫穿Spark的方方面面，包括項目的架構設計、用到的技術的剖析、開發實現、運維等，完整掌握其中的每一個階段和細節，這樣就可以讓您以後可以從容面對絕大多數Spark項目。
6.提供Spark解決方案
徹底掌握Spark框架源碼的每一個細節;
根據不同的業務場景的需要提供Spark在不同場景的下的解決方案;
根據實際需要，在Spark框架基礎上進行二次開發，打造自己的Spark框架;
前面所述的成為Spark高手的六個階段中的第一和第二個階段可以通過自學逐步完成，隨後的三個階段最好是由高手或者專家的指引下一步步完成，最後一個階段，基本上就是到"無招勝有招"的時期，很多東西要用心領悟才能完成。

F. 怎麼在eclipse里正確導入spark2.0.0 的源碼

應該說這個和是不是Spark項目沒什麼關系。

建議你使用intellij idea，在spark目錄下執行"sbt/sbt gen-idea"，會自動生成.idea項目，導入即可。
idea我不熟，還需要做一些其他的插件配置(python, sbt等)和環境設置。

你也可以使用Eclipse看，Eclipse有scala IDE，把Spark項目當maven工程導入。但是子項目之間的依賴會有點問題，會報錯。

推薦使用前者，向Databricks的開發者看齊；我使用的是後者，我直接依賴了編譯好的包就不會報錯了，純讀源碼的話也勉強可以跟蹤和調試。

另外，我也看有的Committer用vim看spark

G. 大家對spark的源碼了解多少，sparkshuffle，調度，sparkstreaming的源碼

流（Streaming），在大數據時代為數據流處理，就像水流一樣，是數據流；既然是數據流處理，就會想到數據的流入、數據的加工、數據的流出。

日常工作、生活中數據來源很多不同的地方。例如：工業時代的汽車製造、監控設備、工業設備會產生很多源數據；信息時代的電商網站、日誌伺服器、社交網路、金融交易系統、黑客攻擊、垃圾郵件、交通監控等；通信時代的手機、平板、智能設備、物聯網等會產生很多實時數據，數據流無處不在。

在大數據時代SparkStreaming能做什麼？

平時用戶都有網上購物的經歷，用戶在網站上進行的各種操作通過Spark Streaming流處理技術可以被監控，用戶的購買愛好、關注度、交易等可以進行行為分析。在金融領域，通過Spark Streaming流處理技術可以對交易量很大的賬號進行監控，防止罪犯洗錢、財產轉移、防欺詐等。在網路安全性方面，黑客攻擊時有發生，通過Spark Streaming流處理技術可以將某類可疑IP進行監控並結合機器學習訓練模型匹配出當前請求是否屬於黑客攻擊。其他方面，如：垃圾郵件監控過濾、交通監控、網路監控、工業設備監控的背後都是Spark Streaming發揮強大流處理的地方。

大數據時代，數據價值一般怎麼定義？

所有沒經過流處理的數據都是無效數據或沒有價值的數據；數據產生之後立即處理產生的價值是最大的，數據放置越久或越滯後其使用價值越低。以前絕大多數電商網站盈利走的是網路流量（即用戶的訪問量），如今，電商網站不僅僅需要關注流量、交易量，更重要的是要通過數據流技術讓電商網站的各種數據流動起來，通過實時流動的數據及時分析、挖掘出各種有價值的數據；比如：對不同交易量的用戶指定用戶畫像，從而提供不同服務質量；准對用戶訪問電商網站板塊愛好及時推薦相關的信息。

SparkStreaming VSHadoopMR：

Spark Streaming是一個准實時流處理框架，而Hadoop MR是一個離線、批處理框架；很顯然，在數據的價值性角度，Spark Streaming完勝於Hadoop MR。

SparkStreaming VS Storm：

Spark Streaming是一個准實時流處理框架，處理響應時間一般以分鍾為單位，也就是說處理實時數據的延遲時間是秒級別的；Storm是一個實時流處理框架，處理響應是毫秒級的。所以在流框架選型方面要看具體業務場景。需要澄清的是現在很多人認為Spark Streaming流處理運行不穩定、數據丟失、事務性支持不好等等，那是因為很多人不會駕馭Spark Streaming及Spark本身。在Spark Streaming流處理的延遲時間方面，Spark定製版本，會將Spark Streaming的延遲從秒級別推進到100毫秒之內甚至更少。

SparkStreaming優點：

1、提供了豐富的API，企業中能快速實現各種復雜的業務邏輯。

2、流入Spark Streaming的數據流通過和機器學習演算法結合，完成機器模擬和圖計算。

3、Spark Streaming基於Spark優秀的血統。

SparkStreaming能不能像Storm一樣，一條一條處理數據？

Storm處理數據的方式是以條為單位來一條一條處理的，而Spark Streaming基於單位時間處理數據的，SparkStreaming能不能像Storm一樣呢？答案是：可以的。

業界一般的做法是Spark Streaming和Kafka搭檔即可達到這種效果,入下圖：

總結：

使用Spark Streaming可以處理各種數據來源類型，如：資料庫、HDFS，伺服器log日誌、網路流，其強大超越了你想像不到的場景，只是很多時候大家不會用，其真正原因是對Spark、spark streaming本身不了解。

H. 深入理解spark核心思想與源碼分析 怎麼樣

SparkSQL主要的推動者是Databricks。提到SparkSQL不得不提的就是Shark。Shark可以理解為Spark社區這邊搞的一個」HiveonSpark」，把Hive的物理執行計劃使用Spark計算引擎去執行。這裡面會有一些問題，Hive社區那邊沒有把物理執行計劃到執行引擎這個步驟抽象出公共API，所以Spark社區這邊要自己維護一個Hive的分支，而且Hive的設計和發展不太會考慮到如何優化Spark的Job。但是前面提到的HiveonSpark卻是和Hive一起發布的，是由Hive社區控制的。所以後來Spark社區就停止了Shark的開發轉向SparkSQL（「坑了」一部分當時信任Shark的人）。SparkSQL是把SQL解析成RDD的transformation和action，而且通過catalyst可以自由、靈活的選擇最優執行方案。對資料庫有深入研究的人就會知道，SQL執行計劃的優化是一個非常重要的環節，SparkSQL在這方面的優勢非常明顯，提供了一個非常靈活、可擴展的架構。但是SparkSQL是基於內存的，元數據放在內存裡面，不適合作為數據倉庫的一部分來使用。所以有了SparkSQL的HiveContext，就是兼容Hive的SparkSQL。它支持HiveQL,HiveMetastore,HiveSerDesandHiveUDFs以及JDBCdriver。這樣看起來很完美，但是實際上也有一些缺點：SparkSQL依賴於Hive的一個snapshot，所以它總是比Hive的發布晚一個版本，很多Hive新的feature和bugfix它就無法包括。而且目前看Spark社區在Spark的thriftserver方面的投入不是很大，所以感覺它不是特別想朝著這個方向發展。還有一個重要的缺點就是SparkSQL目前還不能通過分析SQL來預測這個查詢需要多少資源從而申請對應的資源，所以在共享集群上無法高效地分配資源和調度任務。

I. 如何對Spark 源碼修改後在Eclipse中使用

Eclipse 下開發調試環境的配置
該小節中使用的各項工具分別為：Windows 7+Eclipse Java EE 4.4.2+Scala 2.10.4+Sbt 0.13.8+Maven3.3.3，測試的 Spark 版本為 1.4.0。
1.配置 IDE：
選擇菜單項 Help->Install new software，添加站點 ，選擇安裝 Scala IDE for Eclipse 以及 Scala IDE Plugins。
對於標准版 Eclipse，還需要安裝單獨的 Maven 插件。
出於配置簡便考慮，也可以使用 Scala 官方提供的已將所有依賴打包好的 Scala IDE。
特別的，由於項目本身存在一些錯誤，請先暫時關閉 Project->Build Automatically 的功能以節省時間。
2.下載 Spark 源代碼：
創建空目錄，執行如下語句：git clone
除了使用 git 指令之外，也可以從 Spark 的 Github 頁面下載打包好的源代碼。
3.將源碼轉化為 Eclipse 項目：
進入源代碼根目錄，執行如下語句：sbt eclipse。Sbt 執行期間會下載 Spark 所需要的所有 jar 包，因此該步驟會花費很長的時間。其中有一些 jar 包需要使用網路代理等方法才能下載。
4.導入項目至 Eclipse：
選擇菜單項 File->Import，並選擇 General->Existing Projects into Workspace，項目的根路徑選擇源代碼的根路徑，導入所有項目（共有 25 個）。
5.修改 Scala 版本：
進入 Preference->Scala->Installations，添加機器上安裝的 Scala 2.10.4（選擇 lib 目錄）。由於該版本 Spark（1.4.0）是在 Scala 2.10.4 的環境下編寫的，需要在 Eclipse 中修改項目使用的 Scala 版本。方法為：全選項目，右鍵選擇 Scala->Set the Scala Installation 並選擇相應的 Scala 版本。
6.為 old-deps 項目添加 Scala Library：
右鍵選擇 old-deps 項目，選擇 Scala->Add Scala Library to Build Path。
7.Maven install 以生成 spark-streaming-flume-sink 所需要的類：
首先將源代碼根目錄中的 scalastyle-config.xml 文件復制到 spark-streaming-flume-sink 項目根目錄中，而後在 Eclipse 中打開該項目，右鍵選擇 pom.xml 文件，選擇 Run as->Maven install。
8.修改 spark-sql 與 spark-hive 的包錯誤：
由於源代碼的包設置有錯誤，為此需要將類文件移至正確的包中
對於 spark-sql 項目，分別選擇 src/test/java 中的 test.org.apache.spark.sql 以及 test.org.apache.spark.sql.sources 包中的所有類，右鍵選擇 Refactor->Move，移動至 org.apache.spark.sql 以及 org.apache.spark.sql.sources 包。
對於 spark-hive 項目，分別選擇 src/test/java 中的 test.org.apache.spark.sql.hive 以及 test.org.apache.spark.sql.hive.execution 包中的所有類，移動至 org.apache.spark.sql.hive 以及 org.apache.spark.sql.hive.execution 包。
9.編譯所有項目：
打開 Project->Build Automatically 功能，等待所有項目編譯成功。
10.檢查是否安裝成功：
將 core 項目中的 src->main->resources->org 文件夾拷貝到 examples 項目中的 target->scala-2.10->classes 中。而後執行 examples 項目中的 org.apache.spark.examples.SparkPi 程序，並設置其 jvm 參數為-Dspark.master=local