Fix and support python2.x or python3.9 on yarn.

tracker/dmlc_tracker/kubernetes.py

@@ -11,9 +11,11 @@
 import sys
 import uuid
 import logging
-from kubernetes import client, config
+if "kubernetes" in sys.modules:
+    from kubernetes import client, config
 from . import tracker
-import yaml
+if "yaml" in sys.modules:
+    import yaml
 
 template_volume = {
         "name":""

tracker/dmlc_tracker/launcher.py

@@ -7,7 +7,7 @@
 import sys
 import os
 import subprocess
-from .util import py_str
+from util import py_str
 
 def unzip_archives(ar_list, env):
     for fname in ar_list:
@@ -55,7 +55,7 @@ def main():
         (classpath, _) = subprocess.Popen('%s/bin/hadoop classpath' % hadoop_home,
                                           stdout=subprocess.PIPE, shell=True,
                                           env=os.environ).communicate()
-        classpath = py_str(class_path)
+        classpath = py_str(classpath)
         for f in classpath.split(':'):
             class_path += glob.glob(f)
 
@@ -77,8 +77,9 @@ def main():
     if 'DMLC_JOB_ARCHIVES' in env:
         unzip_archives(env['DMLC_JOB_ARCHIVES'].split(':'), env)
 
-    ret = subprocess.call(args=sys.argv[1:], env=env)
-    sys.exit(ret)
+    fp = subprocess.Popen(args=sys.argv[1:], stdout=subprocess.PIPE, shell=True, env=env)
+    fp.communicate()
+    sys.exit(fp.returncode)
 
 
 if __name__ == '__main__':

tracker/dmlc_tracker/tracker.py

@@ -327,11 +327,11 @@ def run():
         self.thread.start()
 
     def join(self):
-        while self.thread.isAlive():
+        while self.thread.is_alive():
             self.thread.join(100)
 
     def alive(self):
-        return self.thread.isAlive()
+        return self.thread.is_alive()
 
 class PSTracker(object):
     """
@@ -369,7 +369,7 @@ def __init__(self, hostIP, cmd, port=9091, port_end=9999, envs=None):
 
     def join(self):
         if self.cmd is not None:
-            while self.thread.isAlive():
+            while self.thread.is_alive():
                 self.thread.join(100)
 
     def slave_envs(self):
@@ -381,7 +381,7 @@ def slave_envs(self):
 
     def alive(self):
         if self.cmd is not None:
-            return self.thread.isAlive()
+            return self.thread.is_alive()
         else:
             return False
 

tracker/dmlc_tracker/yarn.py

@@ -34,6 +34,7 @@ def yarn_submit(args, nworker, nserver, pass_env):
     YARN_JAR_PATH = os.path.join(args.yarn_app_dir, 'dmlc-yarn.jar')
     curr_path = os.path.dirname(os.path.abspath(os.path.expanduser(__file__)))
     YARN_BOOT_PY = os.path.join(curr_path, 'launcher.py')
+    YARN_UTIL_PY = os.path.join(curr_path, 'util.py')
 
     if not os.path.exists(YARN_JAR_PATH):
         warnings.warn("cannot find \"%s\", I will try to run build" % YARN_JAR_PATH)
@@ -60,6 +61,7 @@ def yarn_submit(args, nworker, nserver, pass_env):
     fset, new_command = opts.get_cache_file_set(args)
     fset.add(YARN_JAR_PATH)
     fset.add(YARN_BOOT_PY)
+    fset.add(YARN_UTIL_PY)
     ar_list = []
 
     for fname in args.archives:
@@ -112,7 +114,7 @@ def yarn_submit(args, nworker, nserver, pass_env):
     def run():
         """internal running function."""
         logging.debug(cmd)
-        subprocess.check_call(cmd, shell=True, env=env)
+        subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, env=env).communicate()
 
     thread = Thread(target=run, args=())
     thread.setDaemon(True)