Currently airflow's DaskExecutor does  not support specifying queues for tasks, due to dask's lack of an explicit queue specification feature. However, this can be reliably mimicked using dask resources ([details here](https://distributed.dask.org/en/latest/resources.html)). So the set up would look something like this:

```
# starting dask worker that can service airflow tasks submitted with queue=queue_name_1 or queue_name_2
$ dask-worker <address> --resources "queue_name_1=inf, queue_name_2=inf"
```
~~(Unfortunately as far as I know you need to provide a finite resource limit for the workers, so you'd need to provide an arbitrarily large limit, but I think it's worth the minor inconvenience to allow a queue functionality in the dask executor.)~~

```
# airflow/executors/dask_executor.py
def execute_async(
    self,
    key: TaskInstanceKey,
    command: CommandType,
    queue: Optional[str] = None,
    executor_config: Optional[Any] = None,
) -> None:

    self.validate_command(command)

    def airflow_run():
        return subprocess.check_call(command, close_fds=True)

    if not self.client:
        raise AirflowException(NOT_STARTED_MESSAGE)

    ################ change made here #################
    resources = None
    if queue:
        resources = {queue: 1}

    future = self.client.submit(airflow_run, pure=False, resources=resources)
    self.futures[future] = key  # type: ignore
```

airflow/executors/dask_executor.py
"""DaskExecutor submits tasks to a Dask Distributed cluster."""
    def __init__(self, cluster_address=None):
        super().__init__(parallelism=0)
        if cluster_address is None:
            cluster_address = conf.get('dask', 'cluster_address')
        if not cluster_address:
            raise ValueError('Please provide a Dask cluster address in airflow.cfg')
        self.cluster_address = cluster_address
        
        self.tls_ca = conf.get('dask', 'tls_ca')
        self.tls_key = conf.get('dask', 'tls_key')
        self.tls_cert = conf.get('dask', 'tls_cert')
        self.client: Optional[Client] = None
        self.futures: Optional[Dict[Future, TaskInstanceKey]] = None
    def start(self) -> None:
        if self.tls_ca or self.tls_key or self.tls_cert:
            security = Security(
                tls_client_key=self.tls_key,
                tls_client_cert=self.tls_cert,
                tls_ca_file=self.tls_ca,
                require_encryption=True,
            )
        else:
            security = None
        self.client = Client(self.cluster_address, security=security)
        self.futures = {}
    def execute_async(