spark源码

‘壹’ 怎么在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 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()}}}

‘叁’ 如何对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的源码，怎么搭阅读和调试的环境

网上搜一下就有，下面转的董西成的博客 （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需要源码编译

因为不同版本的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项目代码

操作系统
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的源码了解多少，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 源码入口类是哪个类

写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
象