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)

'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
******************************************************************************
 ```
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"
     )
 
 

pip install -e .
parse_log_pytest
requirements.txt
pytest-asyncio
apt-get update apt-get install -y gcc
3.7
requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
certifi @ file:///croot/certifi_1671487769961/work/certifi
cffi==1.15.1
click==8.1.8
cloudpickle==2.2.1
cryptography==44.0.2
dask==2022.2.0
-e git+https://github.com/dask/dask-yarn.git@8f5f97418ddf3287f00be2762a195499010bc3c8#egg=dask_yarn
distributed==2022.2.0
exceptiongroup==1.2.2
fsspec==2023.1.0
grpcio==1.62.3
HeapDict==1.0.1
importlib-metadata==6.7.0
iniconfig==2.0.0
Jinja2==3.1.6
locket==1.0.0
MarkupSafe==2.1.5
msgpack==1.0.5
packaging==24.0
partd==1.4.1
pluggy==1.2.0
protobuf==4.24.4
psutil==7.0.0
pycparser==2.21
pytest==7.4.4
pytest-asyncio==0.21.2
PyYAML==6.0.1
skein==0.8.2
sortedcontainers==2.4.0
tblib==2.0.0
tomli==2.0.1
toolz==0.12.1
tornado==6.2
typing_extensions==4.7.1
zict==2.2.0
zipp==3.15.0

name: dask-yarn
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2022.12.7=py37h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=22.3.1=py37h06a4308_0
  - python=3.7.16=h7a1cb2a_0
  - readline=8.2=h5eee18b_0
  - setuptools=65.6.3=py37h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.38.4=py37h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - cffi==1.15.1
      - click==8.1.8
      - cloudpickle==2.2.1
      - cryptography==44.0.2
      - dask==2022.2.0
      - dask-yarn==0.8.0
      - distributed==2022.2.0
      - exceptiongroup==1.2.2
      - fsspec==2023.1.0
      - grpcio==1.62.3
      - heapdict==1.0.1
      - importlib-metadata==6.7.0
      - iniconfig==2.0.0
      - jinja2==3.1.6
      - locket==1.0.0
      - markupsafe==2.1.5
      - msgpack==1.0.5
      - packaging==24.0
      - partd==1.4.1
      - pluggy==1.2.0
      - protobuf==4.24.4
      - psutil==7.0.0
      - pycparser==2.21
      - pytest==7.4.4
      - pytest-asyncio==0.21.2
      - pyyaml==6.0.1
      - skein==0.8.2
      - sortedcontainers==2.4.0
      - tblib==2.0.0
      - tomli==2.0.1
      - toolz==0.12.1
      - tornado==6.2
      - typing-extensions==4.7.1
      - zict==2.2.0
      - zipp==3.15.0
prefix: /opt/conda/envs/dask-yarn

dask_yarn/tests/test_core.py::test_environment_archive
dask_yarn/tests/test_core.py::test_from_specification_errors dask_yarn/tests/test_core.py::test_configuration[remote] dask_yarn/tests/test_core.py::test_configuration[local] dask_yarn/tests/test_core.py::test_make_specification_errors dask_yarn/tests/test_core.py::test_environment_conda[conda://env-name-env-name] dask_yarn/tests/test_core.py::test_environment_conda[conda:///env/path-/env/path] dask_yarn/tests/test_core.py::test_environment_venv dask_yarn/tests/test_core.py::test_environment_python_path dask_yarn/tests/test_core.py::test_make_submit_specification[local] dask_yarn/tests/test_core.py::test_make_submit_specification[remote] dask_yarn/tests/test_core.py::test_environment_relative_paths
dask_yarn/core.py
dask_yarn/core.py