**Apache Airflow version**: v2.0.1

**Environment**:

- **Cloud provider or hardware configuration**:
- bare metal
- **OS** (e.g. from /etc/os-release):
- Ubuntu 20.04.2 LTS (GNU/Linux 5.4.0-65-generic x86_64)
- **Kernel** (e.g. `uname -a`):
- Linux 5.4.0-65-generic #73-Ubuntu SMP Mon Jan 18 17:25:17 UTC 2021 x86_64 x86_64 x86_64 GNU/Linux
- **Install tools**:
- miniconda
- **Others**:
- python 3.7
- hadoop 2.9.2
- spark 2.4.7

**What happened**:
I have two problems:
1. When starts DAG with spark-submit on yarn with deploy="cluster" airflow doesn't track state driver. 
Therefore when yarn fails, the DAG's state remains in "running mode.

2. When I manual stop DAG's job, for example, mark it as "failed", in Yarn cluster the same job is still running. 
This error occurs because empty environment is passed  to subprocess.Popen  and hadoop bin directory doesn't exist in PATH.

ERROR - [Errno 2] No such file or directory: 'yarn': 'yarn'

**What you expected to happen**:
In the first case, move the task to the "failed" state.
In the second case, stopping the task on yarn cluster.
<!-- What do you think went wrong? -->

**How to reproduce it**:
Reproduce the first issue: Start spark_submit on yarn cluster in  deploy="cluster" and master="yarn", then kill task on yarn UI. On airflow task state remains "running".
Reproduce the second issue: Start spark_submit  on yarn cluster in  deploy="cluster" and master="yarn", then manually change this running job's state to "failed", on Hadoop cluster the same job remains in running state.

**Anything else we need to know**:
I propose the following changes to the  to the airflow\providers\apache\hooks\SparkSubmitHook.py:
1. line 201: 
```python
return 'spark://' in self._connection['master'] and self._connection['deploy_mode'] == 'cluster' 
```
change to 
```python        
return ('spark://' in self._connection['master'] or self._connection['master'] == "yarn") and \
               (self._connection['deploy_mode'] == 'cluster')
```
2. line 659
```python
env = None
```
change to 
``` python
env = {**os.environ.copy(), **(self._env if self._env else {})}
```
Applying this patch solved the above issues.
airflow/providers/apache/spark/hooks/spark_submit.py
for the maven coordinates given with 'packages'
    :type repositories: str
    :param total_executor_cores: (Standalone & Mesos only) Total cores for all executors
        (Default: all the available cores on the worker)
    :type total_executor_cores: int
    :param executor_cores: (Standalone, YARN and Kubernetes only) Number of cores per
        executor (Default: 2)
    :type executor_cores: int
    :param executor_memory: Memory per executor (e.g. 1000M, 2G) (Default: 1G)
    :type executor_memory: str
    :param driver_memory: Memory allocated to the driver (e.g. 1000M, 2G) (Default: 1G)
    :type driver_memory: str
    :param keytab: Full path to the file that contains the keytab
    :type keytab: str
    :param principal: The name of the kerberos principal used for keytab
    :type principal: str
    :param proxy_user: User to impersonate when submitting the application
    :type proxy_user: str
    :param name: Name of the job (default airflow-spark)
    :type name: str
    :param num_executors: Number of executors to launch
    :type num_executors: int
    :param status_poll_interval: Seconds to wait between polls of driver status in cluster
        mode (Default: 1)
    :type status_poll_interval: int
    :param application_args: Arguments for the application being submitted
    :type application_args: list
    :param env_vars: Environment variables for spark-submit. It
        supports yarn and k8s mode too.
    :type env_vars: dict