189 8069 5689
学习spark的任何技术前请先正确理解spark，可以参考： 正确理解Spark
创新互联专注为客户提供全方位的互联网综合服务，包含不限于网站设计制作、成都网站制作、高昌网络推广、重庆小程序开发公司、高昌网络营销、高昌企业策划、高昌品牌公关、搜索引擎seo、人物专访、企业宣传片、企业代运营等，从售前售中售后，我们都将竭诚为您服务，您的肯定，是我们最大的嘉奖；创新互联为所有大学生创业者提供高昌建站搭建服务，24小时服务热线：18982081108，官方网址：www.cdcxhl.com
我们知道spark的RDD支持scala api、java api以及python api，我们分别对scala api与java api做了详细的介绍，本文我们将探讨rdd python api是怎么使用py4j来调用scala/java的api的，从而来实现python api的功能。首先我们先介绍下py4j。
一、py4j
py4j是一个使得python可以调用jvm中的对象的类库。看一个py4j官网上的例子：
首先编写一个java程序
package com.twq.javaapi;
import py4j.GatewayServer;
/**
* Created by tangweiqun on 2017/9/22.
*/
public class AdditionApplication {
public int addition(int first, int second) {
return first + second;
}
public static void main(String[] args) {
AdditionApplication app = new AdditionApplication();
// app is now the gateway.entry_point
//启动一个py4j的服务端，python可以连接到这个服务监听的端口，然后调用java的对象及其方法
GatewayServer server = new GatewayServer(app);
server.start();
}
}上面的java代码依赖一个jar包，我们可以通过maven引进来，如下：
net.sf.py4j py4j 0.10.4
我们可以先在ide中启动上面的类AdditionApplication的main方法，将GatewayServer启动起来
然后，我们打开python解释器，执行下面的代码：
>>> from py4j.java_gateway import JavaGateway>>> gateway = JavaGateway() # connect to the JVM>>> random = gateway.jvm.java.util.Random() # create a java.util.Random instance>>> number1 = random.nextInt(10) # call the Random.nextInt method>>> number2 = random.nextInt(10)>>> print(number1,number2)(2, 7)>>> addition_app = gateway.entry_point # get the AdditionApplication instance>>> addition_app.addition(number1,number2) # call the addition method9
上面的python代码依赖py4j，我们可以根据http://www.py4j.org/install.html#install-instructions的方法来安装py4j。从上面可以看出，我们在python中可以很简单的调用jvm中的Random以及AdditionApplication对象的方法
二、py4j在spark中实现python api调用java/scala api
首先，我们编写一个很简单的python版本的spark应用，如下：
if __name__ == "__main__":
conf = SparkConf().setAppName("appName")
sc = SparkContext(conf=conf)
sourceDataRDD = sc.textFile("hdfs://master:9999/users/hadoop-twq/word.txt")
wordsRDD = sourceDataRDD.flatMap(lambda line: line.split())
keyValueWordsRDD = wordsRDD.map(lambda s: (s, 1))
wordCountRDD = keyValueWordsRDD.reduceByKey(lambda a, b: a + b)
wordCountRDD.saveAsTextFile("hdfs://master:9999" + output_path_service.get_output_path())
print utils.get_rdd_result("wordCountRDD", wordCountRDD)上面是一个很简单的python版的spark wordcount应用，我们通过下面的spark-submit命令，提交到spark集群中执行：
spark-submit \ --name "PythonWordCount" \ --master yarn \ --deploy-mode client \ --driver-memory 512m \ --executor-memory 512m \ --num-executors 2 \ --executor-cores 1 \ --py-files word_count_python.zip \ /home/hadoop-twq/spark-course/spark_word_count.py
(对于spark-submit每一个参数的含义以及spark-submit的原理是怎么样的，可以参考：正确提交spark应用)
提交到集群运行后，会在driver端程序启动一个org.apache.spark.deploy.PythonRunner的类，这个里面做了两件事情
1、初始化并启动GatewayServer，如下代码：
// Launch a Py4J gateway server for the process to connect to; this will let it see our
// Java system properties and such
//用于python的代码访问当前jvm中的对象的
val gatewayServer = new py4j.GatewayServer(null, 0)
val thread = new Thread(new Runnable() {
override def run(): Unit = Utils.logUncaughtExceptions {
gatewayServer.start()
}
})
thread.setName("py4j-gateway-init")
thread.setDaemon(true)
thread.start()2、利用ProcessBuilder来启动执行上面的spark_word_count.py python文件，如下：
// Launch Python process
val builder = new ProcessBuilder((Seq(pythonExec, formattedPythonFile) ++ otherArgs).asJava)
val env = builder.environment()
env.put("PYTHONPATH", pythonPath)
// This is equivalent to setting the -u flag; we use it because ipython doesn't support -u:
env.put("PYTHONUNBUFFERED", "YES") // value is needed to be set to a non-empty string
env.put("PYSPARK_GATEWAY_PORT", "" + gatewayServer.getListeningPort)
// pass conf spark.pyspark.python to python process, the only way to pass info to
// python process is through environment variable.
sparkConf.get(PYSPARK_PYTHON).foreach(env.put("PYSPARK_PYTHON", _))
sys.env.get("PYTHONHASHSEED").foreach(env.put("PYTHONHASHSEED", _))
builder.redirectErrorStream(true) // Ugly but needed for stdout and stderr to synchronize这边需要注意的是PYSPARK_GATEWAY_PORT这个环境变量，这个环境变量的值是第一步启动起来的GatewayServer的监听端口，我们将这个端口以环境变量的方式传递给启动的python进程。
然后，当上面的第2步启动了spark_word_count.py python进程后，开始执行spark_word_count.py中的内容，当执行到sc=SparkContext(conf)，即初始化SparkContext，这个时候在SparkContext初始化的时候，会启动一个py4j的Gateway来和上面启动的GatewayServer进行通讯，如下代码(在context.py文件中)：
SparkContext._ensure_initialized(self, gateway=gateway, conf=conf)
def _ensure_initialized(cls, instance=None, gateway=None, conf=None): """ Checks whether a SparkContext is initialized or not. Throws error if a SparkContext is already running. """ with SparkContext._lock: if not SparkContext._gateway: #这里是启动一个Gateway，并且将Gateway的jvm赋值给成员变量_jvm，这样我们就可以通过这个_jvm变量来访问jvm中的java对象及其方法 SparkContext._gateway = gateway or launch_gateway(conf) SparkContext._jvm = SparkContext._gateway.jvm if instance: if (SparkContext._active_spark_context and SparkContext._active_spark_context != instance): currentMaster = SparkContext._active_spark_context.master currentAppName = SparkContext._active_spark_context.appName callsite = SparkContext._active_spark_context._callsite # Raise error if there is already a running Spark context raise ValueError( "Cannot run multiple SparkContexts at once; " "existing SparkContext(app=%s, master=%s)" " created by %s at %s:%s " % (currentAppName, currentMaster, callsite.function, callsite.file, callsite.linenum)) else: SparkContext._active_spark_context = instance
在launch_gateway(conf)(源代码在java_gateway.py中)方法中会初始化一个Gateway，如下：
#从环境变量中拿到环境变量PYSPARK_GATEWAY_PORT，这个就是我们在PythonRunner中设置的环境变量 if "PYSPARK_GATEWAY_PORT" in os.environ: gateway_port = int(os.environ["PYSPARK_GATEWAY_PORT"]) # Connect to the gateway # 启动一个JavaGateway同GatewayServer进行通讯 gateway = JavaGateway(GatewayClient(port=gateway_port), auto_convert=True) #将python api中需要的java/scala的类引入引进来 # Import the classes used by PySpark java_import(gateway.jvm, "org.apache.spark.SparkConf") java_import(gateway.jvm, "org.apache.spark.api.java.*") java_import(gateway.jvm, "org.apache.spark.api.python.*") java_import(gateway.jvm, "org.apache.spark.ml.python.*") java_import(gateway.jvm, "org.apache.spark.mllib.api.python.*") # TODO(davies): move into sql java_import(gateway.jvm, "org.apache.spark.sql.*") java_import(gateway.jvm, "org.apache.spark.sql.hive.*") java_import(gateway.jvm, "scala.Tuple2") return gateway
这样，python中的SparkContext就可以访问RDD java api了，如下是在python文件context.py中访问java api的JavaSparkContext：
def _initialize_context(self, jconf): """ Initialize SparkContext in function to allow subclass specific initialization """ return self._jvm.JavaSparkContext(jconf)