**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
"""
    This hook is a wrapper around the spark-submit binary to kick off a spark-submit job.
    It requires that the "spark-submit" binary is in the PATH or the spark_home to be
    supplied.
    :param conf: Arbitrary Spark configuration properties
    :type conf: dict
    :param conn_id: The connection id as configured in Airflow administration. When an
        invalid connection_id is supplied, it will default to yarn.
    :type conn_id: str
    :param files: Upload additional files to the executor running the job, separated by a
        comma. Files will be placed in the working directory of each executor.
        For example, serialized objects.
    :type files: str
    :param py_files: Additional python files used by the job, can be .zip, .egg or .py.
    :type py_files: str
    :param: archives: Archives that spark should unzip (and possibly tag with #ALIAS) into
        the application working directory.
    :param driver_class_path: Additional, driver-specific, classpath settings.
    :type driver_class_path: str
    :param jars: Submit additional jars to upload and place them in executor classpath.
    :type jars: str
    :param java_class: the main class of the Java application
    :type java_class: str
    :param packages: Comma-separated list of maven coordinates of jars to include on the
        driver and executor classpaths
    :type packages: str
    :param exclude_packages: Comma-separated list of maven coordinates of jars to exclude
        while resolving the dependencies provided in 'packages'
    :type exclude_packages: str
    :param repositories: Comma-separated list of additional remote repositories to search