Ray.distribute_execute中,关注2个配置:
envs : 定义当前用户,python的一些环境,比如SQL中定义的
!python env "PYTHON_ENV=source activate ray1.8.0";
,后面会由ArrowPythonRunner
去传递到python进程里面。runnerConf :保存了HOME、OWNER、GROUP_ID三要素,会通过Input流传输给python进程。
我们会在driver侧启动2个socketServer,分别是:
temp-model-server-in-spark:用于方便python侧读取模型
temp-data-server-in-spark:用于方便python侧读取数据集
上述2个Server都是随机生成的端口。如果5分钟内没有client来连接,会自动关闭掉。
然后我们会看在driver运行还是executor运行,这里启动pythonRunner的方式是一样的,区别是driver模式会使用之前起好的dataServer执行,而executor模式是使用mapPartition在每个executor中运行。
在选择好执行进程后,会创建ArrowPythonRunner
(属于pyjava二方包,即byzer-python)。
我们看下ArrowPythonRunner
如何启动python进程并把参数传递到python中的。
val batch = new ArrowPythonRunner(
Seq(ChainedPythonFunctions(Seq(PythonFunction(
code, envs4j, "python", pythonVersion)))), sourceSchema,
timezoneID, runnerConf
)
我们看下pyjava中的逻辑。
我们会调用父类BasePythonRunner中的compute方法,首先,设置几个参数到envVars(即传入的envs),分别如下:
PY_WORKER_REUSE = 1 : 是否启动的时候重用worker
PY_EXECUTOR_MEMORY:设置python的内存。在用户显式关掉它之前, 他是一个常驻的python worker. 我们可以通过如下方式限制Worker的大小。`!python conf "py_executor_memory=600";``` 实际是从conf中取到放到envs中的
BUFFER_SIZE:输入输出流的buffer设置
值得注意的是,无论!python env
还是!python conf
都是session级别有效的。这意味着一旦设置之后,会影响当前用户后续所有的操作。
我们看下worker是如何启动的,在方法createPythonWorker中,会根据一个Tuple key获取是否有缓存好的worker,代码如下:
def createPythonWorker(pythonExec: String, envVars: Map[String, String], conf: Map[String, String]): java.net.Socket = {
synchronized {
val key = (pythonExec, envVars)
pythonWorkers.getOrElseUpdate(key, new PythonWorkerFactory(pythonExec, envVars, conf)).create()
}
可以看到,如果一个python worker已经创建了,就会被缓存到pythonWorkers中。因为key只有pythonExec,envVars,没有runnerConf,如果用户pythonExec中code和python环境没有变化,就会取到创建好的python worker,所以相应的pythonExec,** envVars,** conf都会被缓存起来不会再变化了,这里就会有相同任务执行多次,GroupId不会变化的bug。因为GroupId是在runnerConf中。
通过create调用创建python进程的逻辑。
pythonWorkers.getOrElseUpdate(key, new PythonWorkerFactory(pythonExec, envVars, conf)).create()
*启动python会有2种模式,一种是直接启动python的worker,第二种是先启动python daemon进程,然后再启动相应的python worker;worker的方式并没有给用户暴露出来,所以默认会走daemon方式。
通过createThroughDaemon创建deamon进程,而如果daemon存在则直接复用,即相同的envs下只有一个daemon进程。因为java中fork进程开销比较大,创建python worker都是在python进程中完成。
byzer进程退出后,daemon会自动销毁。
pyjava会提供一个叫data_manager的变量,方便接受和返回数据给MLSQL主程序。 主要有两个方法: 获取数据, data_manager.fetch_once(), 返回一个迭代器,注意,该方法只能调用一次。 设置返回数据, data_manager.set_output(value) value格式必须是 [[pandas.serial,pandas.serial,...]]