概要

这是一个看前一段时间spark的python支持的时，有点简单的后pyspark内python代码，我们把一个一般流程。虽然几乎没有python，但基本上能看懂pyspark它是如何使不同的虚拟机之间的数据传输、怎么样python通话环境java类别、pyspark SDK的丰富程度取决于什么、须要做些什么流程和封装等。

我看了下，应该仅仅有这篇wiki里介绍了pyspark的实现机制。大体是以下这张图就能够表示：

在python driver端，SparkContext利用Py4J启动一个JVM并产生一个JavaSparkContext。Py4J仅仅使用在driver端，用于本地python与java SparkContext objects的通信。大量数据的传输使用的是还有一个机制。

RDD在python下的转换会被映射成java环境下PythonRDD。在远端worker机器上，PythonRDD对象启动一些子进程并通过pipes与这些子进程通信，以此send用户代码和数据。

大致流程

java_gateway.py里启动了py4j.JavaGateWay。并从java里导入了所须要的主要类。

python能通过py4j訪问jvm的前提是，jvm开启了GatewayServer，而在coreproject的deployproject下，PythonRunner单例里启动了GatewayServer。可能能够理解为py4j是基于socket的一套简单封装了调用java类和方法的协议吧。并且走的本地不同port。

py4j的包为$SPARK_HOME/python/lib/py4j-0.8.1-src.zip。里面是py4j源代码的几个类。

上述java_gateway的launch_gateway()方法是在context.py初始化的时候调用。

context.py初始化的时候，把SparkContext和其部分主要方法增加到了python环境中。所以大多数的调用都是通过py4j直接调用java的类。

java的类主要是指core项目里的java api里的内容。序列化採用了cPickle库的PickleSerializer。

 

像python下使用spark sql的话，在sql.py里，从jvm里获取了SQLContext/HiveContext类，从而得到spark sql里的关键方法。

 

还有一方面，worker.py里，worker启动的时候会起一个socket。从socket里。能够获取工作文件夹名字。能够获取PYTHONPATH下的其它要引入的.zip或.egg文件。将其加到file_dir里。这里的反序列化使用的是UTF8Deserializer；能够获取广播的变量，这里的反序列化使用的是PickleSerializer。

在daemon.py里，通过分配socketport，启动POOLSIZE个worker。(里面还有非常多其它细节)。使用os.fork的方式创建子进程来启动。

 

在rdd.py里。声明了rdd的非常多action和transformations。有些操作会触发数据在python worker上的传输。

传输大量数据的时候。Py4J非常慢。由于socket.readline()非常低效。传输的时候。把数据(序列化后)dump成一个文件。

兴许把这个文件反序列化回来后，能够转成python的类型和结构进行查看和输出(如collect)，也能够调用PythonRDD的asJavaRDD方法(如PipelinedRDD计算时)。在各个worker上启动python进程运行反序列化之后的函数，通过管道与python进程进行通信，最后得到JavaRDD。

 

管道传输利用的是Popen，这样做标准输入

总结

过了一下pyspark是怎么让不同虚拟机之间数据传输并在python环境调用java类的。两者使用的是不同的网络实现在的方法。

掌声 :)

版权声明：本文博客原创文章。博客，未经同意，不得转载。