Apache Spark源码走读（八）Graphx实现剖析&spark repl实现详解

2021-12-30 17:20:03

<一>Graphx实现剖析

概要

图的并行化处理一直是一个非常热门的话题，这里头的重点有两个，一是如何将图的算法并行化，二是找到一个合适的并行化处理框架。Spark作为一个非常优秀的并行处理框架，将一些并行化的算法移到其上面就成了一个很自然的事情。

Graphx是一些图的常用算法在Spark上的并行化实现，同时提供了丰富的API接口。本文就Graphx的代码架构及pagerank在graphx中的具体实现做一个初步的学习。

Google为什么赢得了搜索引擎大战

当Google还在起步的时候，在搜索引擎领域，Yahoo正如日中天，红的发紫。显然，在Google面前的是一堵让人几乎没有任何希望的墙。

但世事难料，现在“外事问谷歌”成了不争的事实，Yahoo应也陪客了。

这种转换到底是如何形成的了，有一个因素是这样的，那就是Google发明了显著提高搜索准确率的PageRank算法。如果说PageRank算法的提出让谷歌牢牢站稳了搜索引擎大战的脚跟，这是毫不夸张的。

搜索引擎有几个要考虑的关键因素(个人观点而已)。

要想吸引用户，就必须要有出色的搜索准确率
有了用户，才能做广告投放，提高广告投放的针对性就可以盈利

上述两个方面都有非常优秀的算法。

废话少述，回到正题。PageRank算法是图论的一个具体应用，ok, 转到图论。

图论简介

图的组成

离散数学中非常重要的一个部分就是图论，下面是一个无向连通图：

顶点(vertex)

上图中的A,B,C,D,E称为图的顶点。

边

顶点与顶点之间的连线称之为边。

图的数学表示

读大学的时候，一直没有想明白为什么要学劳什子的线性代数。直到这两天看《数学之美》一书时，才发觉，线性代数在一些计算机应用领域，那简直就是不可或缺啊。

我们比较容易理解的平面几何和立体几何（一个是二维，一个是三维)，而线性代数解决的其实是一个高维问题，由于无法直觉的感受到，所以很难。如果想比较通俗的理解一下数学为什么有这么多的分支及其内在关联，强烈推荐读一下《数学桥对高等数学的一次观赏之旅》。

在数学中，用什么来表示图呢，答案就是线性代数里面的矩阵，想想看，图的关联矩阵，图的邻接矩阵。总之就是矩阵啦，线性代数一下子有用了。下面是一个具体的例子。

图的并行化处理

刚才说到图可以用矩阵来表示，图的并行化问题在某种程度上就被转化为矩阵运算的并行化问题。

那么以矩阵的乘法为例，看看其是否可以并行化处理。

以矩阵 A X B 为例，说明并行化处理过程。

将上述的矩阵A和B划分为四个部分，如下图所示：

首次对齐之后：

子矩阵相乘

相乘之后，A的子矩阵左移，B的子矩阵上移

计算结果合并

图的并行化处理框架，从Pregel说起

上一节的重点有两点：

图用矩阵来表示，对图的运算就是矩阵的运算
矩阵乘法运算可以并行化，动态演示其并行化的原理

你说ok，我明白了。哪有没有一种合适的并行化处理框架可以用来进行图的计算呢，那你肯定想到了MapReduce。

MapReduce尽管也是一个不错的并行化处理框架，但在图计算方面，有许多缺点，主要是计算的中间过程需要存储到硬盘，效率很低。

Google针对图的并行处理，专门提出了一个了不起的框架Pregel。其执行时的动态视图如下所示。

Pregel有如下优点

级联可扩性好 scalability
容错性强
能够很好的表示各种图的常用算法

Pregel的计算模型

计算模型如下图所示，重要的有三个：

作用于每个顶点的处理逻辑 vertexProgram
消息发送，用于相邻节点间的通讯 sendMessage
消息合并逻辑 messageCombining

Pregel在Spark中的实现

非常感谢你能坚持看到现在，这篇博客内容很多，有点难。我想还是上一幅图将其内在逻辑整一下再继续说下去。

该图要表示的意思是这样的，Graphx利用了Spark这样了一个并行处理框架来实现了图上的一些可并行化执行的算法。

本篇博客要表达的意思就是上面加红的这句话，请诸位看官仔细理解。

算法是否能够并行化与Spark本身无关
算法并行化与否的本身，需要通过数学来证明
已经证明的可并行化算法，利用Spark来实现会是一个错的选择，因为Graphx支持pregel的图计算模型

Graphx中的重要概念

Graph

毫无疑问，图本身是graphx中一个非常重要的概念。

成员变量

graph中重要的成员变量分别为

vertices
edges
triplets

为什么要引入triplets呢，主要是和Pregel这个计算模型相关，在triplets中，同时记录着edge和vertex. 具体代码就不罗列了。

成员函数

函数分成几大类：

对所有顶点或边的操作，但不改变图结构本身，如mapEdges, mapVertices
子图,类似于集合操作中的filter subGraph
图的分割，即paritition操作，这个对于Spark计算来说，很关键，正是因为有了不同的Partition,才有了并行处理的可能, 不同的PartitionStrategy,其收益不同。最容易想到的就是利用Hash来将整个图分成多个区域。
outerJoinVertices 顶点的外连接操作

图的运算和操作 GraphOps

图的常用算法是集中抽象到GraphOps这个类中，在Graph里作了隐式转换，将Graph转换为GraphOps

implicit def graphToGraphOps[VD: ClassTag, ED: ClassTag]
      (g: Graph[VD, ED]): GraphOps[VD, ED] = g.ops

支持的操作如下

collectNeighborIds
collectNeighbors
collectEdges
joinVertices
filter
pickRandomVertex
pregel
pageRank
staticPageRank
connectedComponents
triangleCount
stronglyConnectedComponents

RDD

RDD是Spark体系的核心，那么Graphx中引入了哪些新的RDD呢，有俩，分别为

VertexRDD
EdgeRDD

较之EdgeRdd，VertexRDD更为重要，其上的操作也很多，主要集中于Vertex之上属性的合并，说到合并就不得不扯到关系代数和集合论，所以在VertexRdd中能看到许多类似于sql中的术语，如

leftJoin
innerJoin

至于leftJoin, innerJoin, outerJoin的区别，建议谷歌一下，不再赘述。

Graphx场景分析

图的存储和加载

在进行数学计算的时候，图用线性代数中的矩阵来表示，那么如何进行存储呢？

学数据结构的时候，老师肯定说过好多的办法，不再啰嗦了。

不过在大数据的环境下，如果图很巨大，表示顶点和边的数据不足以放在一个文件中怎么办？ 用HDFS

加载的时候，一台机器的内存不足以容下怎么办？ 延迟加载，在真正需要数据时，将数据分发到不同机器中，采用级联方式。

一般来说，我们会将所有与顶点相关的内容保存在一个文件中vertexFile，所有与边相关的信息保存在另一个文件中edgeFile。

生成某一个具体的图时，用edge就可以表示图中顶点的关联关系，同时图的结构也表示出来了。

GraphLoader

graphLoader是graphx中专门用于图的加载和生成，最重要的函数就是edgeListFile,定义如下。

def edgeListFile(
      sc: SparkContext,
      path: String,
      canonicalOrientation: Boolean = false,
      minEdgePartitions: Int = 1,
      edgeStorageLevel: StorageLevel = StorageLevel.MEMORY_ONLY,
      vertexStorageLevel: StorageLevel = StorageLevel.MEMORY_ONLY)
    : Graph[Int, Int] =
  {
    val startTime = System.currentTimeMillis

    // Parse the edge data table directly into edge partitions
    val lines = sc.textFile(path, minEdgePartitions).coalesce(minEdgePartitions)
    val edges = lines.mapPartitionsWithIndex { (pid, iter) =>
      val builder = new EdgePartitionBuilder[Int, Int]
      iter.foreach { line =>
        if (!line.isEmpty && line(0) != '#') {
          val lineArray = line.split("\\s+")
          if (lineArray.length < 2) {
            logWarning("Invalid line: " + line)
          }
          val srcId = lineArray(0).toLong
          val dstId = lineArray(1).toLong
          if (canonicalOrientation && srcId > dstId) {
            builder.add(dstId, srcId, 1)
          } else {
            builder.add(srcId, dstId, 1)
          }
        }
      }
      Iterator((pid, builder.toEdgePartition))
    }.persist(edgeStorageLevel).setName("GraphLoader.edgeListFile - edges (%s)".format(path))
    edges.count()

    logInfo("It took %d ms to load the edges".format(System.currentTimeMillis - startTime))

    GraphImpl.fromEdgePartitions(edges, defaultVertexAttr = 1, edgeStorageLevel = edgeStorageLevel,
      vertexStorageLevel = vertexStorageLevel)
  } // end of edgeListFile

应用举例之PageRank

什么是PageRank

PageRank是Google专有的算法，用于衡量特定网页相对于搜索引擎索引中的其他网页而言的重要程度。它由Larry Page 和 Sergey Brin在20世纪90年代后期发明。PageRank实现了将链接价值概念作为排名因素。

PageRank将对页面的链接看成投票，指示了重要性。

pageRank的核心思想

”在互联网上，如果一个网页被很多其它网页所链接，说明它受到普遍的承认和依赖，那么它的排名就很高。“ （摘自数学之美第10章）

你说这也太简单了吧，不是跟没说一个样吗，怎么用数学来表示呢？

呵呵，起初我也这么想的，后来多看了几遍之后，明白了一点点。分析步骤用文字表述如下，

网页和网页之间的关系用图来表示
网页A和网页B之间的连接关系表示任意一个用户从网页A到转到网页B的可能性（概率）
所有网页的排名用一维向量来B来表示

所有网页之间的连接用矩阵A来表示，所有网页排名用B来表示。

pageRank如何进行并行化

好了，上面的数学阐述说明了“网页排名的计算可以最终抽象为矩阵相乘”，而在开始的时候已经证明过矩阵相乘可以并行化处理。

理论研究结束了，接下来的就是工程实现了，借用Pregel模型，PageRank中定义的各主要函数分别如下。

vertexProgram

def vertexProgram(id: VertexId, attr: (Double, Double), msgSum: Double): (Double, Double) = {
      val (oldPR, lastDelta) = attr
      val newPR = oldPR + (1.0 - resetProb) * msgSum
      (newPR, newPR - oldPR)
    }

sendMessage

def sendMessage(edge: EdgeTriplet[(Double, Double), Double]) = {
      if (edge.srcAttr._2 > tol) {
        Iterator((edge.dstId, edge.srcAttr._2 * edge.attr))
      } else {
        Iterator.empty
      }
    }

messageCombiner

def messageCombiner(a: Double, b: Double): Double = a + b

一点点启示

通过pageRank这个例子，我们能够搞清楚如何将平素学习的数学理论用以解决实际问题。

“学习的东西总是有价值的，至于用的上用不上，全靠造化了”

完整代码

// Connect to the Spark cluster
val sc = new SparkContext("spark://master.amplab.org", "research")
// Load my user data and parse into tuples of user id and attribute list
val users = (sc.textFile("graphx/data/users.txt")
  .map(line => line.split(",")).map( parts => (parts.head.toLong, parts.tail) ))
// Parse the edge data which is already in userId -> userId format
val followerGraph = GraphLoader.edgeListFile(sc, "graphx/data/followers.txt")
// Attach the user attributes
val graph = followerGraph.outerJoinVertices(users) {
  case (uid, deg, Some(attrList)) => attrList
  // Some users may not have attributes so we set them as empty
  case (uid, deg, None) => Array.empty[String]
}
// Restrict the graph to users with usernames and names
val subgraph = graph.subgraph(vpred = (vid, attr) => attr.size == 2)
// Compute the PageRank
val pagerankGraph = subgraph.pageRank(0.001)
// Get the attributes of the top pagerank users
val userInfoWithPageRank = subgraph.outerJoinVertices(pagerankGraph.vertices) {
  case (uid, attrList, Some(pr)) => (pr, attrList.toList)
  case (uid, attrList, None) => (0.0, attrList.toList)
}

println(userInfoWithPageRank.vertices.top(5)(Ordering.by(_._2._1)).mkString("\n"))

小结

本篇讲来讲去就在强调一个问题，Spark是一个分布式并行计算框架。能不能用Spark，其实大体取决于问题的数学模型本身，如果可以并行化处理，则用之，切不可削足适履。

另一个用张图来总结一下提到的数学知识吧。

再一次强烈推荐《数学桥》

参考资料

《数学之美》
《数学桥高等数学的观赏之旅》
《大数据》

<二>spark repl实现详解

概要

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

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

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

全局视图

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

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

这一部分的内容，解释的非常详细的某过于《深入理解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文件的执行过程大致如下

fork创建一个进程
调用execve来执行ELF
ELF的加载过程中会有动态加载和链接发生
全局变量的初始化，这一部分和glibc相关
执行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 "$@" --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 "$@"

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

SparkSubmit
repl.Main
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中输入的语句是如何被真正执行的呢？

简要的步骤是这样的

在repl中输入的每一行语句，都会被封装为一个object，这一工作主要由interpreter完成
对该object进行编译
由classloader加载编译后的java bytecode
执行引擎负责真正执行加载入内存的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里可以找到, :)

小结

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

参考资料

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

码农公寓

<一>Graphx实现剖析

概要

Google为什么赢得了搜索引擎大战

图论简介

图的组成

顶点(vertex)

边

图的数学表示

图的并行化处理

图的并行化处理框架，从Pregel说起

Pregel的计算模型

Pregel在Spark中的实现

Graphx中的重要概念

Graph

成员变量

成员函数

图的运算和操作 GraphOps

RDD

Graphx场景分析

图的存储和加载

GraphLoader

应用举例之PageRank

什么是PageRank

pageRank的核心思想

pageRank如何进行并行化

vertexProgram

sendMessage

messageCombiner

一点点启示

完整代码

小结

参考资料

<二>spark repl实现详解

概要

全局视图

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

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

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

Java类的加载和执行

序列化和反序列化

spark-shell的执行路径

利用jvisualvm验证

spark repl vs. scala repl

scala repl执行过程

interpreter in scala repl

interpreter in spark repl

IMain.scala vs. SparkIMain.scala

compiler

小结

参考资料

相关文章