Repository: dask/dask-yarn

Issue:
'ZMQIOLoop' object has no attribute 'asyncio_loop'
Hi everyone,

First and foremost, many thanks for your efforts on this project - I'm very excited about it's potential, and have learned a lot just trying to debug this error!

**Issue I'm Seeing**
I am facing the below error when trying to initialize a YarnCluster with dask-yarn.  It happens right off the bat:

`from dask_yarn import YarnCluster
from dask.distributed import Client

cluster = YarnCluster(environment='./py3.tar.gz',
                      worker_vcores=2,
                      worker_memory="8GiB")`

Results in:
`AttributeError: 'ZMQIOLoop' object has no attribute 'asyncio_loop'
`


**Environment**
- **Version information**
Python 3.6.8 (being executed from a CDH CDSW edge node container, if relevant)
Dask-yarn 0.9.0
CDH 6.2.0

From command line, I can submit jobs to my yarn cluster with skein.  I.e., this works:

```
skein driver start
skein application submit test.yaml
```

After running that, I can log into my Yarn application log and see it happilly chugging along (you can also see it via skein application ls on command line, as expected).  While I don't think it will be helpful, I do include the yarn log from this successful job.



**Relevant logs/tracebacks**
***Python***

```
AttributeError                            Traceback (most recent call last)
in engine
      1 cluster = YarnCluster(environment='./py3.tar.gz',
      2                       worker_vcores=2,
----> 3                       worker_memory="8GiB")

/home/cdsw/.local/lib/python3.6/site-packages/dask_yarn/core.py in __init__(self, environment, n_workers, worker_vcores, worker_memory, worker_restarts, worker_env, scheduler_vcores, scheduler_memory, deploy_mode, name, queue, tags, user, host, port, dashboard_address, skein_client, asynchronous, loop)
    390             asynchronous=asynchronous,
    391             loop=loop,
--> 392             skein_client=skein_client,
    393         )
    394 

/home/cdsw/.local/lib/python3.6/site-packages/dask_yarn/core.py in _init_common(self, spec, application_client, host, port, dashboard_address, asynchronous, loop, skein_client)
    533 
    534         if not self.asynchronous:
--> 535             self._sync(self._start_internal())
    536 
    537     def _start_cluster(self):

/home/cdsw/.local/lib/python3.6/site-packages/dask_yarn/core.py in _sync(self, task)
    699         if self.asynchronous:
    700             return task
--> 701         future = asyncio.run_coroutine_threadsafe(task, self.loop.asyncio_loop)
    702         try:
    703             return future.result()

AttributeError: 'ZMQIOLoop' object has no attribute 'asyncio_loop'
```

***Yarn Test with Skein***

 ```
Container: container_e43_1579106210787_0194_01_000001 on **.wm.edu_8041
LogAggregationType: AGGREGATED
***
LogType:container-localizer-syslog
LogLastModifiedTime:Fri Mar 13 15:12:26 +0000 2020
LogLength:1058
LogContents:
2020-03-13 11:11:21,726 INFO [main] org.apache.hadoop.security.authentication.util.KerberosName: Non-simple name ***@campus.wm.edu after auth_to_local rule RULE:[1:$1@$0]/L
2020-03-13 11:11:21,794 INFO [main] org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ContainerLocalizer: Disk Validator: yarn.nodemanager.disk-validator is loaded.
2020-03-13 11:11:21,892 INFO [main] org.apache.hadoop.security.authentication.util.KerberosName: Non-simple name ***@campus.wm.edu after auth_to_local rule RULE:[1:$1@$0]/L
2020-03-13 11:11:21,978 INFO [main] org.apache.hadoop.security.authentication.util.KerberosName: Non-simple name ***@campus.wm.edu after auth_to_local rule RULE:[1:$1@$0]/L
2020-03-13 11:11:22,578 WARN [ContainerLocalizer Downloader] org.apache.hadoop.ipc.Client: Exception encountered while connecting to the server : org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.ipc.StandbyException): Operation category READ is not supported in state standby. Visit https://s.apache.org/sbnn-error

End of LogType:container-localizer-syslog
*******************************************************************************************

Container: container_e43_1579106210787_0194_01_000001 on **.wm.edu_8041
LogAggregationType: AGGREGATED
***
LogType:application.driver.log
LogLastModifiedTime:Fri Mar 13 15:12:26 +0000 2020
LogLength:13
LogContents:
Hello World!
End of LogType:application.driver.log
***************************************************************************************
Container: container_e43_1579106210787_0194_01_000001 on **.wm.edu_8041
LogAggregationType: AGGREGATED
***
LogType:application.master.log
LogLastModifiedTime:Fri Mar 13 15:12:26 +0000 2020
LogLength:1950
LogContents:
20/03/13 11:11:23 INFO skein.ApplicationMaster: Starting Skein version 0.8.0
20/03/13 11:11:23 INFO util.KerberosName: Non-simple name ***@campus.wm.edu after auth_to_local rule RULE:[1:$1@$
0]/L
20/03/13 11:11:23 INFO skein.ApplicationMaster: Running as user ***@campus.wm.edu
20/03/13 11:11:23 INFO conf.Configuration: resource-types.xml not found
20/03/13 11:11:23 INFO resource.ResourceUtils: Unable to find 'resource-types.xml'.
20/03/13 11:11:23 INFO skein.ApplicationMaster: Application specification successfully loaded
20/03/13 11:11:23 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java c
lasses where applicable
20/03/13 11:11:24 INFO skein.ApplicationMaster: gRPC server started at **.wm.edu:35450
20/03/13 11:11:24 INFO skein.ApplicationMaster: WebUI server started at **.wm.edu:36029
20/03/13 11:11:24 INFO skein.ApplicationMaster: Registering application with resource manager
20/03/13 11:11:24 INFO skein.ApplicationMaster: Starting application driver
20/03/13 11:12:24 INFO skein.ApplicationMaster: Shutting down: Application driver completed successfully.
20/03/13 11:12:24 INFO skein.ApplicationMaster: Unregistering application with status SUCCEEDED
20/03/13 11:12:24 INFO impl.AMRMClientImpl: Waiting for application to be successfully unregistered.
20/03/13 11:12:24 WARN ipc.Client: Exception encountered while connecting to the server : org.apache.hadoop.ipc.RemoteExcept
ion(org.apache.hadoop.ipc.StandbyException): Operation category READ is not supported in state standby. Visit https://s.apac
he.org/sbnn-error
20/03/13 11:12:25 INFO skein.ApplicationMaster: Deleted application directory hdfs://nameservice1/user/***@campus
.wm.edu/.skein/application_1579106210787_0194
20/03/13 11:12:25 INFO skein.ApplicationMaster: WebUI server shut down
20/03/13 11:12:25 INFO skein.ApplicationMaster: gRPC server shut down
End of LogType:application.master.log
***************************************************************************************
End of LogType:prelaunch.err
******************************************************************************
Container: container_e43_1579106210787_0194_01_000001 on **.wm.edu_8041
LogAggregationType: AGGREGATED
***
LogType:prelaunch.out
LogLastModifiedTime:Fri Mar 13 15:12:26 +0000 2020
LogLength:70
LogContents:
Setting up env variables
Setting up job resources
Launching container
End of LogType:prelaunch.out
******************************************************************************
 ```

Discussion:

Patch:
diff --git a/dask_yarn/core.py b/dask_yarn/core.py
index 39b18b5..c241bbb 100644
--- a/dask_yarn/core.py
+++ b/dask_yarn/core.py
@@ -122,21 +122,19 @@ def _files_and_build_script(environment):
     if scheme in {"conda", "venv", "python"}:
         path = environment[len(scheme) + 3 :]
         files = {}
+        if scheme == "conda":
+            setup = "conda activate %s" % path
+            cli = "dask-yarn"
+        elif scheme == "venv":
+            setup = "source %s/bin/activate" % path
+            cli = "dask-yarn"
+        else:
+            setup = ""
+            cli = "%s -m dask_yarn.cli" % path
     else:
-        # Treat archived environments the same as venvs
-        scheme = "venv"
-        path = "environment"
         files = {"environment": environment}
-
-    if scheme == "conda":
-        setup = "conda activate %s" % path
-        cli = "dask-yarn"
-    elif scheme == "venv":
-        setup = "source %s/bin/activate" % path
-        cli = "dask-yarn"
-    else:
-        setup = ""
-        cli = "%s -m dask_yarn.cli" % path
+        setup = "source environment/bin/activate"
+        cli = "environment/bin/python -m dask_yarn.cli"
 
     def build_script(cmd):
         command = "%s %s" % (cli, cmd)


Test patch:
diff --git a/dask_yarn/tests/test_core.py b/dask_yarn/tests/test_core.py
index 812077c..82164a6 100644
--- a/dask_yarn/tests/test_core.py
+++ b/dask_yarn/tests/test_core.py
@@ -318,17 +318,16 @@ def test_environment_conda(env, path):
     scheduler = spec.services["dask.scheduler"]
     assert not scheduler.files
     assert (
-        scheduler.script
-        == ("conda activate %s\n" "dask-yarn services scheduler") % path
+        scheduler.script == ("conda activate %s\ndask-yarn services scheduler") % path
     )
     worker = spec.services["dask.worker"]
     assert not worker.files
-    assert worker.script == ("conda activate %s\n" "dask-yarn services worker") % path
+    assert worker.script == ("conda activate %s\ndask-yarn services worker") % path
     client = spec.services["dask.client"]
     assert set(client.files) == {"script.py"}
     assert (
         client.script
-        == ("conda activate %s\n" "dask-yarn services client script.py foo bar") % path
+        == ("conda activate %s\ndask-yarn services client script.py foo bar") % path
     )
 
 
@@ -342,18 +341,16 @@ def test_environment_venv():
     assert not scheduler.files
     assert (
         scheduler.script
-        == ("source %s/bin/activate\n" "dask-yarn services scheduler") % path
+        == ("source %s/bin/activate\ndask-yarn services scheduler") % path
     )
     worker = spec.services["dask.worker"]
     assert not worker.files
-    assert (
-        worker.script == ("source %s/bin/activate\n" "dask-yarn services worker") % path
-    )
+    assert worker.script == ("source %s/bin/activate\ndask-yarn services worker") % path
     client = spec.services["dask.client"]
     assert set(client.files) == {"script.py"}
     assert (
         client.script
-        == ("source %s/bin/activate\n" "dask-yarn services client script.py foo bar")
+        == ("source %s/bin/activate\ndask-yarn services client script.py foo bar")
         % path
     )
 
@@ -383,18 +380,20 @@ def test_environment_archive():
     scheduler = spec.services["dask.scheduler"]
     assert set(scheduler.files) == {"environment"}
     assert scheduler.script == (
-        "source environment/bin/activate\n" "dask-yarn services scheduler"
+        "source environment/bin/activate\n"
+        "environment/bin/python -m dask_yarn.cli services scheduler"
     )
     worker = spec.services["dask.worker"]
     assert set(worker.files) == {"environment"}
     assert worker.script == (
-        "source environment/bin/activate\n" "dask-yarn services worker"
+        "source environment/bin/activate\n"
+        "environment/bin/python -m dask_yarn.cli services worker"
     )
     client = spec.services["dask.client"]
     assert set(client.files) == {"environment", "script.py"}
     assert client.script == (
         "source environment/bin/activate\n"
-        "dask-yarn services client script.py foo bar"
+        "environment/bin/python -m dask_yarn.cli services client script.py foo bar"
     )