Inherit from distributed.Cluster

dask_jobqueue/core.py

@@ -8,6 +8,7 @@
 
 from distributed.utils import tmpfile, ignoring, get_ip_interface, parse_bytes
 from distributed import LocalCluster
+from distributed.deploy import Cluster
 
 dirname = os.path.dirname(sys.executable)
 
@@ -16,7 +17,7 @@
 
 
 @docstrings.get_sectionsf('JobQueueCluster')
-class JobQueueCluster(object):
+class JobQueueCluster(Cluster):
     """ Base class to launch Dask Clusters for Job queues
 
     This class should not be used directly, use inherited class appropriate
@@ -153,10 +154,6 @@ def start_workers(self, n=1):
     def scheduler(self):
         return self.cluster.scheduler
 
-    @property
-    def scheduler_address(self):
-        return self.cluster.scheduler_address
-
     def _calls(self, cmds):
         """ Call a command using subprocess.communicate
 
@@ -223,15 +220,3 @@ def __enter__(self):
     def __exit__(self, type, value, traceback):
         self.stop_workers(self.jobs)
         self.cluster.__exit__(type, value, traceback)
-
-    def adapt(self):
-        """ Start up an Adaptive deployment if not already started
-
-        This makes the cluster request resources in accordance to current
-        demand on the scheduler """
-        from distributed.deploy import Adaptive
-        if self._adaptive:
-            return
-        else:
-            self._adaptive = Adaptive(self.scheduler, self, startup_cost=5,
-                                      key=lambda ws: ws.host)

requirements.txt

@@ -0,0 +1,2 @@
+distributed>=1.21.3
+docrep

setup.py

@@ -9,6 +9,6 @@
       url='https://github.com/dask/dask-jobqueue',
       license='BSD 3-Clause',
       packages=['dask_jobqueue'],
+      install_requires=open('requirements.txt').read().strip().split('\n'),
       long_description=(open('README.rst').read() if exists('README.rst') else ''),
-      install_requires=['docrep'],
       zip_safe=False)