Support Calling Java Function in Python Executor and ModelBroadcast in Python #2284

yangw1234 · 2018-02-07T12:11:22Z

What changes were proposed in this pull request?

Support Calling Java Function in Python Executor and ModelBroadcast in Python

To enable this feature, user should explicitly call

init_executor_gateway(sc)

The approach we took is

launch a JavaGateway server on each executor and write the listening port to a fine in the executor's working directory (typical the ${SPARK_HOMW}/work/app-id/executor-id)
in python, if running on executor, find the port on the above file and create a gateway client to connect the executor JVM.

How was this patch tested?

add a unit test to for spark local mode
manually tested on spark standalone cluster

zhichao-li · 2018-02-08T02:08:08Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/utils/Engine.scala

+  private var gatewayServer: py4j.GatewayServer = null
+
+
+  private[bigdl] def createJavaGateway(driverPort: Int): Unit = {


should it be a singleton here? i.e createJavaGateway is invoked multiple times on the same Executor.

On java side, Engine is already a singleton, so it won't create multiple servers.

zhichao-li · 2018-02-08T02:15:27Z

pyspark/bigdl/util/common.py

@@ -610,9 +634,14 @@ def _java2py(sc, r, encoding="bytes"):

 def callJavaFunc(sc, func, *args):
    """ Call Java Function """
-    args = [_py2java(sc, a) for a in args]
+    if sc is None:


Move this checking into GatewayWrapper ?

if run_on_driver: return get_spark_context()._jvm else: return create_gateway()....

zhichao-li · 2018-02-08T02:16:16Z

pyspark/bigdl/util/common.py

@@ -533,6 +541,8 @@ def get_spark_context(conf=None):
    :param conf: combining bigdl configs into spark conf
    :return: SparkContext
    """
+    if SparkFiles._is_running_on_worker:


remove this?

zhichao-li · 2018-02-08T02:24:44Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/utils/Engine.scala

+    }
+    val thread = new Thread(new Runnable() {
+      override def run(): Unit = try {
+        gatewayServer.start()


should we add a hooker to stop the gatewayServer once the thread is closed? i.e:
Runtime.getRuntime().addShutdownHook

yiheng · 2018-02-08T05:46:26Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/utils/Engine.scala

+
+  private[bigdl] def createJavaGateway(driverPort: Int): Unit = {
+    if (SparkUtils.isDriver) {
+      val file = new java.io.File(SparkFiles.getRootDirectory(), "gateway_port")


Should we make sure it's thread safe when check and write the file?

yiheng · 2018-02-08T05:48:32Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/utils/Engine.scala

+      if (file.exists()) {
+        file.delete()
+      }
+      file.createNewFile()


Please add some logs when initializing gateway server, makes it easier to monitor

yiheng · 2018-02-08T05:49:05Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/utils/Engine.scala

+        file.delete()
+      }
+      file.createNewFile()
+      val out = new PrintWriter(file)


should we try to catch the file create exception and throw a more meaningful error message?

yiheng · 2018-02-08T05:49:37Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/utils/Engine.scala

+    }
+    val thread = new Thread(new Runnable() {
+      override def run(): Unit = try {
+        gatewayServer.start()


If the gatewayServer crashs, will it output some log and where can we find it?

I turned on logging, logs can be found on spark's executor's log file.

yiheng · 2018-02-08T05:49:59Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/utils/Engine.scala

+        case ct: ControlThrowable =>
+          throw ct
+        case t: Throwable =>
+          throw new Exception(s"Uncaught exception in thread ${Thread.currentThread().getName}", t)


Please replace with a more meaning error message

yiheng · 2018-02-08T05:53:42Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/utils/Engine.scala

+      }
+    })
+
+    val file = new java.io.File(SparkFiles.getRootDirectory(), "gateway_port")


should it be thread safe?

yiheng · 2018-02-08T05:54:04Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/utils/Engine.scala

+    if (file.exists()) {
+      file.delete()
+    }
+    file.createNewFile()


try to catch the exceptions and throw a more meaningful error

yiheng · 2018-02-08T05:56:31Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/utils/Engine.scala

+      if (gatewayServer != null) return
+      gatewayServer = new py4j.GatewayServer(null, 0)
+    }
+    val thread = new Thread(new Runnable() {


what if another thread try to use this gateway while the gateway havn't inited

Only one thread can reach this part of code.

yiheng · 2018-02-08T05:57:21Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/python/api/PythonBigDL.scala

@@ -2940,6 +2935,11 @@ class PythonBigDL[T: ClassTag](implicit ev: TensorNumeric[T]) extends Serializab
    maxIteration: Int): SequentialSchedule = {
    seq.add(scheduler, maxIteration)
  }
+
+  def initExecutorGateway(sc: JavaSparkContext, driverPort: Int): Unit = {


private bigdl?

yiheng · 2018-02-08T05:57:48Z

pyspark/bigdl/util/common.py

@@ -438,6 +443,9 @@ def uniform(self, a, b, size):
 def init_engine(bigdl_type="float"):
    callBigDlFunc(bigdl_type, "initEngine")

+def init_executor_gateway(sc, bigdl_type="float"):


when call this method?

yiheng · 2018-02-08T06:00:05Z

In each iteration, between the env and agent, there will be a tensor serial and deserial, right?

yangw1234 · 2018-02-08T06:33:44Z

@yiheng yes, but I can't think of any way around this.

zhichao-li · 2018-02-08T08:44:32Z

pyspark/test/bigdl/test_simple_integration.py

@@ -533,7 +533,7 @@ def test_save_jtensor_dict(self):
        tensors["tensor1"] = JTensor.from_ndarray(np.random.rand(3, 2))
        tensors["tensor2"] = JTensor.from_ndarray(np.random.rand(3, 2))
        # in old impl, this will throw an exception
-        _py2java(self.sc, tensors)
+        _py2java(self.sc._gateway, tensors)


add unittest for the new added ModelBroadcast?

zhichao-li · 2018-02-09T01:23:28Z

pyspark/bigdl/models/utils/model_broadcast.py

+from bigdl.nn.layer import Model
+
+def _from_id_and_type(bid, bigdl_type):
+    from pyspark.broadcast import _broadcastRegistry


call method _from_id directly to avoid duplicating the code here?

zhichao-li · 2018-02-09T01:58:38Z

LGTM

yangw1234 · 2018-02-09T02:22:50Z

PR Validation pass

jason-dai · 2018-02-12T13:51:35Z

pyspark/bigdl/util/common.py

+def _get_port():
+    root_dir = SparkFiles.getRootDirectory()
+    path = os.path.join(root_dir, "gateway_port")
+    f = open(path)


what if this fails?

I'll submit another PR and make it report a meaningful error message.

jason-dai · 2018-02-12T14:11:06Z

spark/dl/src/main/scala/com/intel/analytics/bigdl/utils/Engine.scala

+
+  private[bigdl] def createJavaGateway(driverPort: Int): Unit = {
+    if (SparkUtils.isDriver) {
+      if (driverPortFileCreated.compareAndSet(false, true)) {


Why is it needed on driver?

In spark local mode, python worker is a different process from python driver, so it cannot get the SparkContext object and need this to know the driver gateway's listening port.

…n Python (intel-analytics#2284) * support call java function in executor * fix test * fix style * address comments * add parallelism * address comments * fix partition num * address comments * fix typo * fix typo * add integration test * address comments

* add hyperzoo for k8s support (intel-analytics#2140) * add hyperzoo for k8s support * format * format * format * format * run examples on k8s readme (intel-analytics#2163) * k8s readme * fix jdk download issue (intel-analytics#2219) * add doc for submit jupyter notebook and cluster serving to k8s (intel-analytics#2221) * add hyperzoo doc * add hyperzoo doc * add hyperzoo doc * add hyperzoo doc * fix jdk download issue (intel-analytics#2223) * bump to 0.9s (intel-analytics#2227) * update jdk download url (intel-analytics#2259) * update some previous docs (intel-analytics#2284) * K8docsupdate (intel-analytics#2306) * Update README.md * Update s3 related links in readme and documents (intel-analytics#2489) * Update s3 related links in readme and documents * Update s3 related links in readme and documents * Update s3 related links in readme and documents * Update s3 related links in readme and documents * Update s3 related links in readme and documents * Update s3 related links in readme and documents * update * update * modify line length limit * update * Update mxnet-mkl version in hyper-zoo dockerfile (intel-analytics#2720) Co-authored-by: gaoping <pingx.gao@intel.com> * update bigdl version (intel-analytics#2743) * update bigdl version * hyperzoo dockerfile add cluster-serving (intel-analytics#2731) * hyperzoo dockerfile add cluster-serving * update * update * update * update jdk url * update jdk url * update Co-authored-by: gaoping <pingx.gao@intel.com> * Support init_spark_on_k8s (intel-analytics#2813) * initial * fix * code refactor * bug fix * update docker * style * add conda to docker image (intel-analytics#2894) * add conda to docker image * Update Dockerfile * Update Dockerfile Co-authored-by: glorysdj <glorysdj@gmail.com> * Fix code blocks indents in .md files (intel-analytics#2978) * Fix code blocks indents in .md files Previously a lot of the code blocks in markdown files were horribly indented with bad white spaces in the beginning of lines. Users can't just select, copy, paste, and run (in the case of python). I have fixed all these, so there is no longer any code block with bad white space at the beginning of the lines. It would be nice if you could try to make sure in future commits that all code blocks are properly indented inside and have the right amount of white space in the beginning! * Fix small style issue * Fix indents * Fix indent and add \ for multiline commands Change indent from 3 spaces to 4, and add "\" for multiline bash commands Co-authored-by: Yifan Zhu <fanzhuyifan@gmail.com> * enable bigdl 0.12 (intel-analytics#3101) * switch to bigdl 0.12 * Hyperzoo example ref (intel-analytics#3143) * specify pip version to fix oserror 0 of proxy (intel-analytics#3165) * Bigdl0.12.1 (intel-analytics#3155) * bigdl 0.12.1 * bump 0.10.0-Snapshot (intel-analytics#3237) * update runtime image name (intel-analytics#3250) * update jdk download url (intel-analytics#3316) * update jdk8 url (intel-analytics#3411) Co-authored-by: ardaci <dongjie.shi@intel.com> * update hyperzoo docker image (intel-analytics#3429) * update hyperzoo image (intel-analytics#3457) * fix jdk in az docker (intel-analytics#3478) * fix jdk in az docker * fix jdk for hyperzoo * fix jdk in jenkins docker * fix jdk in cluster serving docker * fix jdk * fix readme * update python dep to fit cnvrg (intel-analytics#3486) * update ray version doc (intel-analytics#3568) * fix deploy hyperzoo issue (intel-analytics#3574) Co-authored-by: gaoping <pingx.gao@intel.com> * add spark fix and net-tools and status check (intel-analytics#3742) * intsall netstat and add check status * add spark fix for graphene * bigdl 0.12.2 (intel-analytics#3780) * bump to 0.11-S and fix version issues except ipynb * add multi-stage build Dockerfile (intel-analytics#3916) * add multi-stage build Dockerfile * multi-stage build dockerfile * multi-stage build dockerfile * Rename Dockerfile.multi to Dockerfile * delete Dockerfile.multi * remove comments, add TINI_VERSION to common arg, remove Dockerfile.multi * multi-stage add tf_slim Co-authored-by: shaojie <shaojiex.bai@intel.com> * update hyperzoo doc and k8s doc (intel-analytics#3959) * update userguide of k8s * update k8s guide * update hyperzoo doc * Update k8s.md add note * Update k8s.md add note * Update k8s.md update notes * fix 4087 issue (intel-analytics#4097) Co-authored-by: shaojie <shaojiex.bai@intel.com> * fixed 4086 and 4083 issues (intel-analytics#4098) Co-authored-by: shaojie <shaojiex.bai@intel.com> * Reduce image size (intel-analytics#4132) * Reduce Dockerfile size 1. del redis stage 2. del flink stage 3. del conda & exclude some python packages 4. add copies layer stage * update numpy version to 1.18.1 Co-authored-by: zzti-bsj <shaojiex.bai@intel.com> * update hyperzoo image (intel-analytics#4250) Co-authored-by: Adria777 <Adria777@github.com> * bigdl 0.13 (intel-analytics#4210) * bigdl 0.13 * update * print exception * pyspark2.4.6 * update release PyPI script * update * flip snapshot-0.12.0 and spark2.4.6 (intel-analytics#4254) * s-0.12.0 master * Update __init__.py * Update python.md * fix docker issues due to version update (intel-analytics#4280) * fix docker issues * fix docker issues * update Dockerfile to support spark 3.1.2 && 2.4.6 (intel-analytics#4436) Co-authored-by: shaojie <otnw_bsj@163.com> * update hyperzoo, add lib for tf2 (intel-analytics#4614) * delete tf 1.15.0 (intel-analytics#4719) Co-authored-by: Le-Zheng <30695225+Le-Zheng@users.noreply.github.com> Co-authored-by: pinggao18 <44043817+pinggao18@users.noreply.github.com> Co-authored-by: pinggao187 <44044110+pinggao187@users.noreply.github.com> Co-authored-by: gaoping <pingx.gao@intel.com> Co-authored-by: Kai Huang <huangkaivision@gmail.com> Co-authored-by: GavinGu07 <55721214+GavinGu07@users.noreply.github.com> Co-authored-by: Yifan Zhu <zhuyifan@stanford.edu> Co-authored-by: Yifan Zhu <fanzhuyifan@gmail.com> Co-authored-by: Song Jiaming <litchy233@gmail.com> Co-authored-by: ardaci <dongjie.shi@intel.com> Co-authored-by: Yang Wang <yang3.wang@intel.com> Co-authored-by: zzti-bsj <2779090360@qq.com> Co-authored-by: shaojie <shaojiex.bai@intel.com> Co-authored-by: Lingqi Su <33695124+Adria777@users.noreply.github.com> Co-authored-by: Adria777 <Adria777@github.com> Co-authored-by: shaojie <otnw_bsj@163.com>

yangw1234 requested a review from zhichao-li February 7, 2018 12:11

yiheng self-requested a review February 7, 2018 14:09

zhichao-li reviewed Feb 8, 2018

View reviewed changes

yiheng reviewed Feb 8, 2018

View reviewed changes

zhichao-li reviewed Feb 8, 2018

View reviewed changes

zhichao-li reviewed Feb 9, 2018

View reviewed changes

yangw1234 added 10 commits February 9, 2018 09:45

support call java function in executor

bb38cc2

fix test

c81a36e

fix style

b141b53

address comments

1af41ae

add parallelism

d851be4

address comments

cf3e1c2

fix partition num

9ef1545

address comments

b142440

fix typo

aad927e

fix typo

e1c4ffe

yangw1234 added 2 commits February 9, 2018 09:45

add integration test

b9c3e5a

address comments

e6658a2

yangw1234 force-pushed the py2java branch from 55ce8d2 to e6658a2 Compare February 9, 2018 01:51

yangw1234 merged commit cd8db33 into intel-analytics:master Feb 9, 2018

jason-dai reviewed Feb 12, 2018

View reviewed changes

Le-Zheng added a commit to Le-Zheng/BigDL that referenced this pull request Oct 20, 2021

update some previous docs (intel-analytics#2284)

132d22e

dding3 pushed a commit to dding3/BigDL that referenced this pull request Nov 17, 2021

update some previous docs (intel-analytics#2284)

8f66926

yangw1234 deleted the py2java branch July 20, 2023 17:36

		private var gatewayServer: py4j.GatewayServer = null


		private[bigdl] def createJavaGateway(driverPort: Int): Unit = {

Support Calling Java Function in Python Executor and ModelBroadcast in Python #2284

Support Calling Java Function in Python Executor and ModelBroadcast in Python #2284

Conversation

yangw1234 commented Feb 7, 2018 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yiheng commented Feb 8, 2018 • edited Loading

yangw1234 commented Feb 8, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhichao-li commented Feb 9, 2018

yangw1234 commented Feb 9, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yangw1234 commented Feb 7, 2018 •

edited

Loading

yiheng commented Feb 8, 2018 •

edited

Loading