Spark源码解析 - Spark-shell浅析

1.准备工作

1.1 安装spark，并配置spark-env.sh

使用spark-shell前需要安装spark，详情可以参考http://www.cnblogs.com/swordfall/p/7903678.html

如果只用一个节点，可以不用配置slaves文件，spark-env.sh文件只需配置为master_ip和local_ip两个属性

spark-env.sh添加如下配置：

export SPARK_MASTER_IP=hadoop1
export SPARK_LOCAL_IP=hadoop1

注意：hadoop1是这台虚拟机的ip地址，或者用127.0.0.1代替hadoop1也行。spark-shell浅析是基于spark-2.2.0-bin-hadoop2.7版本进行的。

1.2 启动spark-shell

输入spark安装目录的bin下，执行spark-shell命令

cd /opt/app/spark-2.2.0-bin-hadoop2.7/bin/
./spark-shell

最后我们会看到spark启动的过程，如图所示：

 

2. 执行word count 范例

通过word count例子来感受下spark任务的执行过程，启动spark-shell后，会打开scala命令行，然后按照以下步骤输入脚本。

1) 输入val lines = sc.textFile("../README.md", 2)

2) 输入val words = lines.flatMap(line => line.split(" "))

3) 输入val ones = words.map(w => (w, 1))

4) 输入val counts = ones.reduceByKey(_ + _)

5) 输入counts.foreach(println)

3. 剖析spark-shell

通过word count在spark-shell中执行的过程，看看spark-shell做了什么。spark-shell中有以下一段脚本

我们看到脚本spark-shell里执行了spark-submit脚本，打开spark-submit脚本，发现其中包含以下脚本：

脚本spark-submit在执行spark-class脚本时，给它增加了参数SparkSubmit。打开spark-class脚本，其中包含以下脚本：

读到这里，可知spark-class里面首先加载spark-env.sh里面的配置属性，然后获取jdk的java命令，接着拿到spark_home的jars目录。至此，Spark启动了以SparkSubmit为主类的jvm进程。

为便于在本地对Spark进程使用远程监控，给SPARK_HOME目录conf/spark-defaults.conf配置文件追加以下jmx配置：

#driver端监控
spark.driver.extraJavaOptions=-XX:+UnlockCommercialFeatures -XX:+FlightRecorder -Dcom.sun.management.jmxremote 
-Dcom.sun.management.jmxremote.port=10207 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false

#executor端监控，暂时注释
#spark.executor.extraJavaOptions=-XX:+UnlockCommercialFeatures -XX:+FlightRecorder -Dcom.sun.management.jmxremote 
-Dcom.sun.management.jmxremote.port=0 -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false

注意：使用远程监控前，如果spark-shell在运行中需要先停止，配置好监控参数，再需要运行spark-shell命令，否则jvisualvm找不到该线程。上面的两条命令都是一行的，不是两行，两行会导致jvisualvm连接不上，报“无法使用 service:jmx:rmi:///jndi/rmi://192.168.187.201:8009/jmxrmi 连接到 192.168.187.201:8009”错误。

在本地JAVA_HOME/bin目录下打开jvisualvm，添加远程主机，如图；右击已添加的远程主机，添加JMX连接，如图：

 

 

单击右侧的“线程”选项卡，选择main线程，然后单击“线程Dump”按钮，如图：

从dump的内容中找到线程main的信息，如图：

 

main线程dump信息

   

从main线程的栈信息中可以看出程序的调用顺序：SparkSubmit.main -> repl.Main -> ILoop.process。org.apache.spark.repl.SparkILoop类继承ILoop类，ILoop的process方法调用SparkILoop的loadFiles(settings)与printWelcome()方法。SparkILoop的loadFiles(settings)方法中又调用了自身的initializeSpark方法，initializeSpark的实现如下：

initializationCommands是一个命令集合，见代码：

从代码中可以看到，命令集合中会调用org.apache.spark.repl.Main的createSparkSession()方法创建或者获取sparkSession类，如图：

从上述代码可以看到builder是SparkSession里面的属性，IDEA工具使用“ctrl+鼠标点击”操作，可以进入到builder.getOrCreate()方法里面查看SparkSession如何创建，如图：

 从上述代码可以看到SparkContext首先创建，再创建SparkSession。SparkContext的创建代码如下：

这里使用SparkConf、SparkContext和SparkSession来完成初始化，代码分析中涉及的repl主要用于与Spark实时交互。

4.Spark-shell的整体流程

至此，Spark-shell解析完毕。

参考资料：

《深入理解Spark核心思想与源码分析》

https://www.iteblog.com/archives/1349.html  使用jvisualvm监控Spark作业