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。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来完成初始化,代码分析中涉及的repl主要用于与Spark实时交互。
至此,Spark-shell解析完毕。