Apache Spark源码走读之16 -- spark repl实现详解

欢迎转载，转载请注明出处,徽沪一郎。

概要

之所以对spark shell的内部实现产生兴趣全部缘于好奇代码的编译加载过程，scala是需要编译才能执行的语言，但提供的scala repl可以实现代码的实时交互式执行，这是为什么呢？

既然scala已经提供了repl，为什么spark还要自己单独搞一套spark repl，这其中的缘由到底何在？

显然，这些都是问题，要解开这些谜团，只有再次开启一段源码分析之旅了。

全局视图

上图显示了java源文件从编译到加载执行的全局视图，整个过程中最主要的步骤是

  1. 编译成过程，由编译器对java源文件进行编译整理，生成java bytecodes
  2. 类的加载和初始化，主要由classloader参与
  3. 执行引擎 将字节码翻译成机器码，然后调度执行

这一部分的内容，解释的非常详细的某过于《深入理解jvm》和撒迦的JVM分享，这里就不班门弄斧了。

那么讲上述这些内容的目的又何在呢，我们知道scala也是需要编译执行的，那么编译的结果是什么样呢，要符合什么标准？在哪里执行。

答案比较明显，scala源文件也需要编译成java bytecodes，和java的编译结果必须符合同一份标准，生成的bytecode都是由jvm的执行引擎转换成为机器码之后调度执行。

也就是说尽管scala和java源文件的编译器不同，但它们生成的结果必须符合同一标准，否则jvm无法正确理解，执行也就无从谈起。至于scala的编译器是如何实现的，文中后续章节会涉及。

ELF可执行文件的加载和运行

”CPU是很傻的，加电后，它就会一直不断的读取指令，执行指令，不能停的哦。“ 如果有了这个意识，看源码的时候你就会有无穷的疑惑，无数想不通的地方，这也能让你不断的进步。

再继续讲scala源文件的编译细节之前，我们还是来温习一下基础的内容，即一个EFL可执行文件是如何加载到内存真正运行起来的。(本篇博客的内容相对比较底层，很费脑子的，:)

Linux平台上基本采用ELF作为可执行文件的格式，java可执行文件本身也是ELF格式的，使用file指令来作检验。

file /opt/java/bin/java

下面是输出的结果，从结果中可以证实java也是ELF格式。

/opt/java/bin/java: ELF 64-bit LSB executable, x86-64, version 1 (SYSV), dynamically linked (uses shared libs), for GNU/Linux 2.6.9, BuildID[sha1]=bd74b7294ebbdd93e9ef3b729e5aab228a3f681b, stripped

ELF文件的执行过程大致如下

  1. fork创建一个进程
  2. 调用execve来执行ELF
  3. ELF的加载过程中会有动态加载和链接发生
  4. 全局变量的初始化，这一部分和glibc相关
  5. 执行main函数

我讲述的过程非常笼统，要想更清楚的了解细节，请参阅《深入理解Linux内核》中的程序的执行一章，或是《深入Linux内核架构》中的启动新程序一节。

现在打开内核中相应的源码，看看execve函数是如何找到elf格式的处理句柄的。

第一步：每一种二进制格式，必须先注册自己的处理句柄。

在文件$KERNEL_HOME/fs/binfmt_elf.c中，init_elf_binfmt函数就实现了注册任务

static int __init init_elf_binfmt(void)
{
	register_binfmt(&elf_format);
	return 0;
}

来看一看elf_format的定义是什么

static struct linux_binfmt elf_format = {
	.module		= THIS_MODULE,
	.load_binary	= load_elf_binary,
	.load_shlib	= load_elf_library,
	.core_dump	= elf_core_dump,
	.min_coredump	= ELF_EXEC_PAGESIZE,
};

第二步：搜索处理句柄，fs/exec.c

execve是一个系统调用，内核中对应的函数是do_execve，具体代码不再列出。

do_execve->do_execve_common->search_binary_hander

注意search_binary_handler会找到上一步中注册的binary_handler即elf_format，找到了对应的handler之后，关键的一步就是load_binary了。动态链接过程调用的是load_shlib，这一部分的内容细细展开的话，够写几本书了。

search_binary_handler的部分代码

retry:
	read_lock(&binfmt_lock);
	list_for_each_entry(fmt, &formats, lh) {
		if (!try_module_get(fmt->module))
			continue;
		read_unlock(&binfmt_lock);
		bprm->recursion_depth++;
		retval = fmt->load_binary(bprm);
		bprm->recursion_depth--;
		if (retval >= 0 || retval != -ENOEXEC ||
		    bprm->mm == NULL || bprm->file == NULL) {
			put_binfmt(fmt);
			return retval;
		}
		read_lock(&binfmt_lock);
		put_binfmt(fmt);
	}
	read_unlock(&binfmt_lock);

要想对这一部分内容有个比较清楚的了解，建议看一下台湾黄敬群先生的《深入浅出Helloworld》和国内出版的《程序员的自我修养》。

另外一个值得推荐的是黑客级的网站phrack.org，可惜现在不更新了。

之所以讲ELF的加载和运行，是因为要打通java源文件的编译执行过程的话，必然会步步深入到此，其实到这还不够，再往下走就是CPU指令，只有到达CPU指令才算真正到底。这个时候就需要去读intel ia-64 software programmer guide了。

源码走读其实只是个形式，重要的是能理清楚其执行流程，以到达指令级的理解为最佳。

Java类的加载和执行

在各位java达人面前，我就不显示自己java水平有多烂了。只是将两幅最基本的图搬出来，展示一下java类的加载过程，以及classloader的层次关系。记住这些东东会为我们在后头讨论scala repl奠定良好基础。

序列化和反序列化

Java体系中，另一个重要的基石就是类的序列化和反序列化。这里要注意的就是当有继承体系时，类的序列化和反序列化顺序，以及类中有静态成员变量的时候，如何处理序列化。诸如此类的文章，一搜一大把，我再多加解释实在是画蛇添足，列出来只是说明其重要性罢了。

spark-shell的执行路径

前面进行了这么多的铺垫之后，我想可以进入正题了。即spark-shell的执行调用路径到底怎样。

首次使用Spark一般都是从执行spark-shell开始的,当在键盘上敲入spark-shell并回车时,后面究竟发生了哪些事情呢?

export SPARK_SUBMIT_OPTS
$FWDIR /bin/spark - submit spark -shell "[email protected]" --class org.apache.spark.repl.Main

可以看出spark-shell其实是对spark-submit的一层封装,但事情到这还没有结束,毕竟还没有找到调用java的地方,继续往下搜索看看spark-submit脚本的内容。

exec $SPARK_HOME /bin/spark -class org. apache .spark.
deploy . SparkSubmit "${ ORIG_ARGS [@]}"

离目标越来越近了,spark-class中会调用到java程序,与java相关部分的代码摘录如下

# Find the java binary
if [ -n "${ JAVA_HOME }" ]; then
RUNNER ="${ JAVA_HOME }/ bin/java"
else
if [ `command -v java ` ]; then
RUNNER ="java"
else
echo " JAVA_HOME is not set" >&2
exit 1
fi
fi
exec " $RUNNER " -cp " $CLASSPATH " $JAVA_OPTS "[email protected]"

SparkSubmit当中定义了Main函数,在它的处理中会将spark repl运行起来,spark repl能够接收用户的输入,通过编译与运行,返回结果给用户。这就是为什么spark具有交互处理能力的原因所在。调用顺序如下

  1. SparkSubmit
  2. repl.Main
  3. SparkILoop

利用jvisualvm验证

修改spark-class,使得JAVA_OPTS看起来如下图所示

JMX_OPTS="-Dcom.sun.management.jmxremote.port=8300 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=127.0.0.1"
# Set JAVA_OPTS to be able to load native libraries and to set heap size
JAVA_OPTS="-XX:MaxPermSize=128m $OUR_JAVA_OPTS $JMX_OPTS"
JAVA_OPTS="$JAVA_OPTS -Xms$OUR_JAVA_MEM -Xmx$OUR_JAVA_MEM"

修改完上述脚本之后先启动spark-shell,然后再启动jvisualvm

bin/spark-shell
jvisualvm

在Java VisualVM中选择进程org.apache.spark.deploy.SparkSubmit，如果已经为jvisualvm安装了插件Threads Inspector，其界面将会与下图很类似

在右侧选择“线程”这一tab页，选择线程main,然后可以看到该线程的thread dump信息

spark repl vs. scala repl

既然scala已经提供了repl, spark还是要自己去实现一个repl，你不觉着事有可疑么？我谷歌了好长时间，终于找到了大神的讨论帖子，不容易啊,原文摘录如下。

Thanks for looping me in! Just FYI, I would also be okay if instead of making the wrapper code pluggable, the REPL just changed to one based on classes, as in Prashant‘s example, rather than singleton objects.

 

To give you background on this, the problem with the "object" wrappers is that initialization code goes into a static initializer that will have to run on all worker nodes, making the REPL unusable with distributed applications. As an example, consider this:

 

// file.txt is a local file on just the master

val data = scala.io.Source.fromFile("file.txt").mkString

 

// now we use the derived string, "data", in a closure that runs on the cluster

spark.textFile.map(line => doStuff(line, data))

 

The current Scala REPL creates an object Line1 whose static initializer sets data with the code above, then does import Line1.data in the closure, which will cause the static initializer to run *again* on the remote node and fail. This issue definitely affects Spark, but it could also affect other interesting projects that could be built on Scala‘s REPL, so it may be an interesting thing to consider supporting in the standard interpreter.

 

Matei

上述内容估计第一次看了之后，除了一头雾水还是一头雾水。翻译成为白话就是利用scala原生的repl，是使用object来封装输入的代码的，这有什么不妥，“序列化和反序列化”的问题啊。反序列化的过程中，对象的构造函数会被再次调用，而这并不是我们所期望的。我们希望生成class而不是object，如果你不知道object和class的区别，没关系，看一下scala的简明手册，马上就明白了。

最重要的一点：Scala Repl默认输入的代码都是在本地执行，故使用objectbasedwraper是没有问题的。但在spark环境下，输入的内容有可能需要在远程执行，这样objectbasedwrapper的源码生成方式经序列化反序列化会有相应的副作用，导致出错不可用。

讨论详情，请参考该Link https://groups.google.com/forum/#!msg/scala-internals/h27CFLoJXjE/JoobM6NiUMQJ

scala repl执行过程

再啰嗦一次,scala是需要编译执行的，而repl给我们的错觉是scala是解释执行的。那我们在repl中输入的语句是如何被真正执行的呢？

简要的步骤是这样的

  1. 在repl中输入的每一行语句，都会被封装为一个object， 这一工作主要由interpreter完成
  2. 对该object进行编译
  3. 由classloader加载编译后的java bytecode
  4. 执行引擎负责真正执行加载入内存的bytecode

interpreter in scala repl

那么怎么证明我说的是对的呢？很简单，做个实验，利用下述语句了启动scala repl

scala -Dscala.repl.debug=true

如果我们输入这样一条语句 val c = 10，由interpreter生成的scala源码会如下所列

object $read extends scala.AnyRef {
  def () = {
    super.;
    ()
  };
  object $iw extends scala.AnyRef {
    def () = {
      super.;
      ()
    };
    object $iw extends scala.AnyRef {
      def () = {
        super.;
        ()
      };
      val c = 10
    }
  }
}

注意啰，是object哦，不是class。

interpreter in spark repl

那我们再看看spark repl生成的scala源码是什么样子的？

启动spark-shell之前，修改一下spark-class，在JAVA_OPTS中加入如下内容

-Dscala.repl.debug=true

启动spark-shell,输入val b = 10,生成的scala源码如下所示

class $read extends AnyRef with Serializable {
    def (): $line10.$read = {
      $read.super.();
      ()
    };
    class $iwC extends AnyRef with Serializable {
      def (): $read.this.$iwC = {
        $iwC.super.();
        ()
      };
      class $iwC extends AnyRef with Serializable {
        def (): $iwC = {
          $iwC.super.();
          ()
        };
        import org.apache.spark.SparkContext._;
        class $iwC extends AnyRef with Serializable {
          def (): $iwC = {
            $iwC.super.();
            ()
          };
          class $iwC extends AnyRef with Serializable {
            def (): $iwC = {
              $iwC.super.();
              ()
            };
            private[this] val b: Int = 100;
              def b: Int = $iwC.this.b
          };
          private[this] val $iw: $iwC = new $iwC.this.$iwC();
            def $iw: $iwC = $iwC.this.$iw
        };
        private[this] val $iw: $iwC = new $iwC.this.$iwC();
          def $iw: $iwC = $iwC.this.$iw
      };
      private[this] val $iw: $iwC = new $iwC.this.$iwC();
        def $iw: $iwC = $iwC.this.$iw
    };
    private[this] val $iw: $read.this.$iwC = new $read.this.$iwC();
      def $iw: $read.this.$iwC = $read.this.$iw
  };
  object $read extends scala.AnyRef with Serializable {
    def (): $line10.$read.type = {
      $read.super.();
      ()
    };
    private[this] val INSTANCE: $line10.$read = new $read();
      def INSTANCE: $line10.$read = $read.this.INSTANCE;
     private def readResolve(): Object = $line10.this.$read
  }
}

注意到与scala repl中的差异了么，此处是class而非object

IMain.scala vs. SparkIMain.scala

是什么导致有上述的差异的呢？我们可以下载scala的源码，对是scala本身的源码在github上可以找到。interpreter中代码生成部分的处理逻辑主要是在IMain.scala，在spark中是SparkIMain.scala。

比较两个文件的异同。

gvimdiff IMain.scala SparkIMain.scala

gvimdiff是个好工具，两个文件的差异一目了然，emacs和vim总要有一样玩的转才行啊。来个屏幕截图吧，比较炫吧。

注：spark开发团队似乎给scala的开发小组提了一个case，在最新的scala中似乎已经支持classbasedwrapper，可以通过现应的选项来设置来选择classbasedwraper和objectbasedwrapper.

下述代码见最新版scala，scala-2.12.x中的IMain.scala

  private lazy val ObjectSourceCode: Wrapper =
      if (settings.Yreplclassbased) new ClassBasedWrapper else new ObjectBasedWrapper

compiler

scala实现了自己的编译器，处理逻辑的代码实现见scala源码中的src/compiler目录下的源文件。有关其处理步骤不再赘述，请参考ref3,ref4中的描述。

有一点想要作个小小提醒的时，当你看到SparkIMain.scala中有new Run的语句却不知道这个Run在哪的时候，兄弟跟你讲在scala中的Global.scala里可以找到, :)

小结

编译和加载是一个非常有意思的话题，即可以说是很基础也可以说很冷门，有无动力就这部分进行深究，就看个人的兴趣了。

参考资料

  1. http://phrack.org/issues/58/5.html
  2. http://blog.linux.org.tw/~jserv/archives/002125.html
  3. http://lampwww.epfl.ch/~magarcia/ScalaCompilerCornerReloaded/2012Q2/GenASM.pdf
  4. https://wiki.scala-lang.org/display/SIW/Overview+of+Compiler+Phases
  5. 《深入理解Linux内核》
  6. 《深入Linux内核架构》
  7. 《深入理解Java虚拟机》

Apache Spark源码走读之16 -- spark repl实现详解

时间： 2024-12-09 23:28:53

Apache Spark源码走读之16 -- spark repl实现详解的相关文章

Apache Spark源码走读之9 -- Spark源码编译

欢迎转载,转载请注明出处,徽沪一郎. 概要 本来源码编译没有什么可说的,对于java项目来说,只要会点maven或ant的简单命令,依葫芦画瓢,一下子就ok了.但到了Spark上面,事情似乎不这么简单,按照spark officical document上的来做,总会出现这样或那样的编译错误,让人懊恼不已. 今天闲来无事,又重试了一把,居然o了,做个记录,以备后用. 准备 我的编译机器上安装的Linux是archlinux,并安装后如下软件 scala 2.11 maven git 下载源码 第

Apache Spark源码走读之22 -- Spark MLLib中拟牛顿法L-BFGS的源码实现

欢迎转载,转载请注明出处,徽沪一郎. 概要 本文就拟牛顿法L-BFGS的由来做一个简要的回顾,然后就其在spark mllib中的实现进行源码走读. 拟牛顿法 数学原理 代码实现 L-BFGS算法中使用到的正则化方法是SquaredL2Updater. 算法实现上使用到了由scalanlp的成员项目breeze库中的BreezeLBFGS函数,mllib中自定义了BreezeLBFGS所需要的DiffFunctions. runLBFGS函数的源码实现如下 def runLBFGS( data:

Apache Spark源码走读之1 -- Spark论文阅读笔记

转自:http://www.cnblogs.com/hseagle/p/3664933.html 楔子 源码阅读是一件非常容易的事,也是一件非常难的事.容易的是代码就在那里,一打开就可以看到.难的是要通过代码明白作者当初为什么要这样设计,设计之初要解决的主要问题是什么. 在对Spark的源码进行具体的走读之前,如果想要快速对Spark的有一个整体性的认识,阅读Matei Zaharia做的Spark论文是一个非常不错的选择. 在阅读该论文的基础之上,再结合Spark作者在2012 Develop

Apache Spark源码走读之8 -- Spark on Yarn

欢迎转载,转载请注明出处,徽沪一郎. 概要 Hadoop2中的Yarn是一个分布式计算资源的管理平台,由于其有极好的模型抽象,非常有可能成为分布式计算资源管理的事实标准.其主要职责将是分布式计算集群的管理,集群中计算资源的管理与分配. Yarn为应用程序开发提供了比较好的实现标准,Spark支持Yarn部署,本文将就Spark如何实现在Yarn平台上的部署作比较详尽的分析. Spark Standalone部署模式回顾 上图是Spark Standalone Cluster中计算模块的简要示意,

Apache Spark源码走读之12 -- Hive on Spark运行环境搭建

欢迎转载,转载请注明出处,徽沪一郎. 楔子 Hive是基于Hadoop的开源数据仓库工具,提供了类似于SQL的HiveQL语言,使得上层的数据分析人员不用知道太多MapReduce的知识就能对存储于Hdfs中的海量数据进行分析.由于这一特性而收到广泛的欢迎. Hive的整体框架中有一个重要的模块是执行模块,这一部分是用Hadoop中MapReduce计算框架来实现,因而在处理速度上不是非常令人满意.由于Spark出色的处理速度,有人已经成功将HiveQL的执行利用Spark来运行,这就是已经非常

Apache Spark源码走读之13 -- hiveql on spark实现详解

欢迎转载,转载请注明出处,徽沪一郎 概要 在新近发布的spark 1.0中新加了sql的模块,更为引人注意的是对hive中的hiveql也提供了良好的支持,作为一个源码分析控,了解一下spark是如何完成对hql的支持是一件非常有趣的事情. Hive简介 Hive的由来 以下部分摘自Hadoop definite guide中的Hive一章 "Hive由Facebook出品,其设计之初目的是让精通SQL技能的分析师能够对Facebook存放在HDFS上的大规模数据集进行分析和查询. Hive大大

Apache Spark源码走读之21 -- 浅谈mllib中线性回归的算法实现

欢迎转载,转载请注明出处,徽沪一郎. 概要 本文简要描述线性回归算法在Spark MLLib中的具体实现,涉及线性回归算法本身及线性回归并行处理的理论基础,然后对代码实现部分进行走读. 线性回归模型 机器学习算法是的主要目的是找到最能够对数据做出合理解释的模型,这个模型是假设函数,一步步的推导基本遵循这样的思路 假设函数 为了找到最好的假设函数,需要找到合理的评估标准,一般来说使用损失函数来做为评估标准 根据损失函数推出目标函数 现在问题转换成为如何找到目标函数的最优解,也就是目标函数的最优化

Apache Spark源码走读之14 -- Graphx实现剖析

欢迎转载,转载请注明出处,徽沪一郎. 概要 图的并行化处理一直是一个非常热门的话题,这里头的重点有两个,一是如何将图的算法并行化,二是找到一个合适的并行化处理框架.Spark作为一个非常优秀的并行处理框架,将一些并行化的算法移到其上面就成了一个很自然的事情. Graphx是一些图的常用算法在Spark上的并行化实现,同时提供了丰富的API接口.本文就Graphx的代码架构及pagerank在graphx中的具体实现做一个初步的学习. Google为什么赢得了搜索引擎大战 当Google还在起步的

Apache Spark源码走读之5 -- DStream处理的容错性分析

欢迎转载,转载请注明出处,徽沪一郎,谢谢. 在流数据的处理过程中,为了保证处理结果的可信度(不能多算,也不能漏算),需要做到对所有的输入数据有且仅有一次处理.在Spark Streaming的处理机制中,不能多算,比较容易理解.那么它又是如何作到即使数据处理结点被重启,在重启之后这些数据也会被再次处理呢? 环境搭建 为了有一个感性的认识,先运行一下简单的Spark Streaming示例.首先确认已经安装了openbsd-netcat. 运行netcatnc -lk 9999 运行spark-s