spark源碼

㈠ 怎麼用Eclipse搭建Spark源碼閱讀環境

第一部分、軟體安裝

1、 安裝JDK （版本為1.7.0_11）

2、 安裝Scala （版本為2.11.2）

3、 安裝ScalaIDE（版本為3.0.4）

第二部分：加壓縮官網下載的源代碼包或者找到通過Git抽取的Spark源文件：

我用的是spark-1.1.1版本（最新版本），由於idea 13已經原生支持sbt，所以無須為idea安裝sbt插件。

源碼下載(用git工具)：

# Masterdevelopment branch

gitclone git://github.com/apache/spark.git

# 1.1 maintenancebranch with stability fixes on top of Spark 1.1.1

gitclone git://github.com/apache/spark.git -b branch-1.1

源碼更新（用git工具同步跟新源碼）：

gitclone https://github.com/apache/spark.git

第三部分：通過sbt工具，構建Scala的Eclipse工程，詳細步驟如下所示

1、通過cmd命令進入DOS界面，之後通過cd命令進入源代碼項目中，我下載的Spark.1.1.1版本的源代碼放在（E:\Spark計算框架的研究\spark_1_1_1_eclipse）文件夾中，之後運行sbt命令，如下所示：

2、運行sbt命令之後，解析編譯相關的jar包，並出現sbt命令界面窗口，出現的效果圖如下所示，之後運行eclipse命令，sbt對這個工程進行編譯，構建Eclipse項目，效果圖如下所示：

4、 打開ScalaIDE工具，File à Import à Existing Projects into Workspace à
Next à
選擇剛好用sbt工具編譯好的Eclipse工程（E:\Spark計算框架的研究\spark_1_1_1_eclipse），如下圖所示。

5、 通過上面的操作，就可以將通過sbt工具編譯生成的Eclipse項目導入到EclipseIDE開發環境中，效果圖如下所示：

錯誤提示如下所示：我導入的包為，如下文件夾中所示。

（E:\Spark計算框架的研究\spark_1_1_1_eclipse\lib_managed\bundles）

Description Resource Path Location Type

akka-remote_2.10-2.2.3-shaded-protobuf.jar is cross-compiled

with an incompatible version of Scala (2.10).

In case of errorneous report, this check can be disabled

in the compiler preference page.

spark-core Unknown Scala Classpath Problem

Description Resource Path Location Type

akka-slf4j_2.10-2.2.3-shaded-protobuf.jar is cross-compiled with

an incompatible version of Scala (2.10). In case of errorneous report,

this check can be disabled in the compiler preference page.

spark-core Unknown Scala Classpath Problem

Description Resource Path Location Type

akka-testkit_2.10-2.2.3-shaded-protobuf.jar is cross-compiled

with an incompatible version of Scala (2.10).

In case of errorneous report, this check can be disabled in the compiler preference page.

spark-core Unknown Scala Classpath Problem

Description Resource Path Location Type

akka-zeromq_2.10-2.2.3-shaded-protobuf.jar is cross-compiled

with an incompatible version of Scala (2.10).

In case of errorneous report, this check can be disabled in the compiler preference page.

spark-core Unknown Scala Classpath Problem

上面這些包兼容性問題還沒有解決，修改相應的jar包就可以解決。

㈡ 怎麼在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

㈢ 大家對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本身不了解。

㈣ 想研讀下spark的源碼，怎麼搭閱讀和調試的環境

網上搜一下就有，下面轉的董西成的博客 （1）准備工作 1） 安裝JDK 6或者JDK 7 2） 安裝scala 2.10.x (注意版本) 3） 將下載的Intellij IDEA解壓後，安裝scala插件，流程如下： 依次選擇「Configure」–> 「Plugins」–> 「Browse repositories」，輸入scala，然後安裝即可 （2）搭建Spark源碼閱讀環境（需要聯網） 第一種方法是直接依次選擇「import project」–> 選擇spark所在目錄 –> 「SBT」，之後intellij會自動識別SBT文件，並下載依賴的外部jar包，整個流程用時非常長，取決於機器的網路環境（不建議在windows下操作，可能遇到各種問題），一般需花費幾十分鍾到幾個小時。注意，下載過程會用到git，因此應該事先安裝了git。 第二種方法是首先在linux操作系統上生成intellij項目文件，然後在intellij IDEA中直接通過「Open Project」打開項目即可。在linux上生成intellij項目文件的方法（需要安裝git，不需要安裝scala，sbt會自動下載）是：在spark源代碼根目錄下，輸入sbt/sbt gen-idea 註：如果你在windows下閱讀源代碼，建議先在linux下生成項目文件，然後導入到windows中的intellij IDEA中。 （3）搭建Spark開發環境 在intellij IDEA中創建scala project，並依次選擇「File」–> 「project structure」 –> 「Libraries」，選擇「+」，將spark-hadoop 對應的包導入，比如導入spark-assembly_2.10-0.9.0-incubating-hadoop2.2.0.jar（只需導入該jar包，其他不需要），如果IDE沒有識別scala 庫，則需要以同樣方式將scala庫導入。之後開發scala程序即可： 編寫完scala程序後，可以直接在intellij中，以local模式運行，方法如下： 點擊「Run」–> 「Run Configurations」，在彈出的框中對應欄中填寫「local」，表示將該參數傳遞給main函數，如下圖所示，之後點擊「Run」–> 「Run」運行程序即可。 如果想把程序打成jar包，通過命令行的形式運行在spark 集群中，可以按照以下步驟操作： 依次選擇「File」–> 「Project Structure」 –> 「Artifact」，選擇「+」–> 「Jar」 –> 「From Moles with dependencies」，選擇main函數，並在彈出框中選擇輸出jar位置，並選擇「OK」。 最後依次選擇「Build」–> 「Build Artifact」編譯生成jar包。

㈤ spark 源碼入口類是哪個類

寫Linux腳本
#!/bin/sh
if [ $# -lt 2 ] ;then
echo "Useage:$0 $1 "
exit
else
echo "wordcount is starting!"
spark-submit \
--master spark://spark:7077 \
--class WordCount \
--executor-memory 512m \
--total-executor-cores 6 \
~/input/WordCount.jar \
"$1" \
"$2"
echo "wordcount is stopped!"
fi
象

㈥ 如何對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

㈦ 如何構建第一個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集群，即可看到運行結果。

㈧ 新手請教：為什麼安裝spark需要源碼編譯

因為不同版本的HDFS在協議上是不兼容的，所以如果你想用你的Spark從HDFS上讀取數據，那麼你就的選擇相應版本的HDFS來編譯Spark，這個可以在編譯的時候通過設置hadoop.version來選擇，默認情況下，Spark默認為編譯成Hadoop 1.0.4版本。現在可以使用的方法有Maven編譯，sbt編譯（時間較長）,Spark自帶腳本編譯（實際還是調用Maven）。

㈨ 怎麼在Idea IDE裡面打開Spark源碼而不報錯

首先我們先點擊一個工程的Project Structure菜單，這時候會彈出一個對話框，仔細的用戶肯定會發現裡面列出來的模塊(Mole)居然沒有yarn！就是這個原因導致yarn模塊相關的代碼老是報錯！只需要將yarn模塊加入到這里即可。
步驟依次選擇 Add->Import Mole->選擇pom.xml，然後一步一步點擊確定，這時候會在對話框裡面多了spark-yarn_2.10模塊，

然後點擊Maven Projects裡面的Reimport All Maven Projects，等yarn模塊裡面的所有依賴全部下載完的時候，我們就可以看到這個模塊裡面的代碼終於不再報錯了！！

㈩ 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()}}}