Deploying dask on YARN in an enterprise setting

*Apologies in advance for the long issue, this is in response to several offline conversations with @mrocklin and @martindurant*

With [skein](https://github.com/jcrist/skein) we have a way to startup arbitrary services on YARN clusters. The intent of that project was to be generally useful, but specifically to get dask working nicely on YARN. This is now doable per-user by ssh'ing into the edge node and running things manually (described in more detail below). The next step is figuring out how this might work within a larger enterprise ecosystem.

This raises a few issues due to the following

- Most Hadoop clusters have a limited number of ports open. IT is responsible for managing these ports, and are usually reluctant to add more unless they're for a specific service that's always running and is known to be secure.
- Because of this, users usually ssh into the *edge node*. From here they can start jobs and do work, interacting with processes that run throughout the cluster.
- Most users have limited disk space on the edge node (<= 200 MB). Larger files should be kept on hdfs, or on external workstations.
- Ideally users should not be running large processes on the edge node. This includes things like Jupyter kernels, the dask scheduler, etc...
- Hadoop security (kerberos, etc...) is complicated and can vary between deployments. Integrating nicely in all cases may be tricky.

---

Below I enumerate 4 options for deploying dask on YARN, increasing in complexity, and discuss their pros and cons.

## 1. Manual Remote kernel


In this configuration, the user does the following:

- SSH into the edge node
- Uses skein to launch a YARN job that starts up a jupyter kernel, dask scheduler, and some number of dask workers
- Uses skein to get the address of the jupyter kernel/dask dashboard/whatever other web stuff is needed
- Uses ssh to tunnel the notebook/dashboard outside the cluster (e.g. `ssh -L external_port:address:port`)
- Interacts with the remote kernel from their browser

This is currently doable using skein and dask. Skein/dask-yarn/whatever could make this process easier on the user by adding support for a reverse-proxy (something like https://github.com/jupyterhub/nbserverproxy) which could handle finding all the dynamic address automatically. This would allow them to always tunnel through the same localhost:port combo:

```
# On the edge node
dask-yarn proxyserver --port 8586 # port unique per user
```

Pros
- Easy, already works
- Requires no IT support
- Flexible

Cons
- Client environment is also remote, must be bundled up the same as all the worker environments.
- Manual process, not as easy as other deployments
- Requires per-user process on edge node

## 2. Manual Local Kernel


This is similar to 1., but runs the client process locally instead of remotely.

- SSH into the edge node
- Uses skein to launch a YARN job that starts up a dask scheduler and some number of dask workers on the cluster
- Uses skein to get the address of the dask scheduler and dashboard
- Uses ssh to tunnel the scheduler and dashboard outside the cluster (e.g. `ssh -L external_port:address:port`)
- Starts a local jupyter kernel
- Uses the local jupyter kernel to start a client and connect to the scheduler over the ssh tunnel

As with 1, this could be made easier by having the user start a small proxy service on the edge node to handle finding the dynamic addresses, so they can always tunnel through the same localhost:port combo.

Pros:
- Easy, already works
- Client environment is local environment
- Requires no IT support
- Flexible

Cons
- Manual process, not as easy as other deployments
- Requires per-user process on edge node

## 3. Jupyter Gateway


This would use https://github.com/jupyter-incubator/enterprise_gateway and jupyterhub to deploy remote kernels as a service. I *think* (but haven't attempted) that adding deployment using skein should be straightforward with their pluggable design.

In this situation, the user would login to jupyterhub, specify what environment they want to deploy, and then click "start notebook". This would start a remote kernel on the cluster, managed by a skein application master. The skein specification that started that kernel could also include other services like dask, which could be scaled up or down by the user easily from inside the notebook using the skein library (already works fine).

Pros:
- Uses existing technology
- The pluggable security nature of jupyterhub makes login management easier (we are not security experts)
- Single application deployed on an edge node, users don't interact with edge node at all
- Only requires one port to be opened on the cluster
- The traffic being proxied is lighter (browser -> kernel) compared to that between the dask client and dask scheduler.

Cons:
- Requires IT support and privileged account to deploy
- Client environment is also remote, must be bundled up the same as all the worker environments.
- Doesn't (currently) play well with other enterprise solutions like AE5

## 4. Dask Gateway

In this case users would start a jupyter kernel anywhere (locally, in AE5, wherever), and then contact a dask specific server on the edge node to spin up a cluster and manage communications for them. This is perhaps the nicest solution, but also the most complicated. It would look something like:

- A single server process running on the edge node, communicating outside the cluster via a single (or a couple) open ports.
- Users would ask the server to spin up a dask cluster for them using YARN with certain files/environments distributed to each worker. This could use Skein and mostly already works (delegation not done yet, but easily addable).
- The server would then need to proxy traffic from the client outside the cluster to the specific scheduler address inside the cluster. I *think* this would require additions/changes to the dask client->scheduler protocol. This is trickier than proxying HTTP, as we can't use the path in the HTTP request to do the dispatching, the information on what scheduler to forward to will need to be part of the TCP message and would be dask specific.
- The server would also need to proxy the dask dashboard outside the cluster. This could use an already built http proxy, or something custom.
- Since we'd be exposing TCP connections to the schedulers outside the cluster, we'd need to include some security mechanism to prevent access to clusters between users. Contrasting with HTTP security for remote kernels, which has many pre-baked solutions, this would need to be something custom. I'd probably design this to have the client/scheduler/workers pass around an authentication token that was created after authenticating with the dask gateway (which could use pluggable security a. la. jupyterhub)

Pros
- Client environment is local environment
- Single application deployed on an edge node, users don't interact with edge node at all
- Only requires one port to be opened on the cluster
- Probably the nicest end solution overall

Cons:
- Requires IT support and privileged account to deploy
- Trickier, requires lots of custom code and security handling
- Traffic being proxied (client -> scheduler) is heavier than browser to notebook