### Body

Spun out of https://github.com/apache/airflow/pull/19665#discussion_r760469246

**Info:**
Currently operators in Amazon provider has a structure of operator per file ([link](https://github.com/apache/airflow/tree/35be8bdee0cdd3e5c73270b0b65e0552fb9d9946/airflow/providers/amazon/aws/operators)) A single service S3 has 7 operators files! This makes it very hard to discover classes.
The proposed structure is what we have for GCP ([link](https://github.com/apache/airflow/tree/35be8bdee0cdd3e5c73270b0b65e0552fb9d9946/airflow/providers/google/cloud/operators)) where file is created for a service and all operators for that services are in a single file.


**Proposed changes:**


**Operators:**

files that don't require changes:

athena.py
batch.py
cloud_formation.py
datasync.py
ecs.py
eks.py
glacier.py
glue.py
glue_crawler.py
redshift.py
sns.py
sqs.py

files that require changes:

|                Current                	|        New       	|
|:-------------------------------------:	|:----------------:	|
|           dms_create_task.py          	|      dms.py      	|
|           dms_delete_task.py          	|      dms.py      	|
|         dms_describe_tasks.py         	|      dms.py      	|
|           dms_start_task.py           	|      dms.py      	|
|            dms_stop_task.py           	|      dms.py      	|
|         ec2_start_instance.py         	|      ec2.py      	|
|          ec2_stop_instance.py         	|      ec2.py      	|
|            emr_add_steps.py           	|      emr.py      	|
|           emr_containers.py           	|      emr.py      	|
|         emr_create_job_flow.py        	|      emr.py      	|
|         emr_modify_cluster.py         	|      emr.py      	|
|       emr_terminate_job_flow.py       	|      emr.py      	|
|              s3_bucket.py             	|       s3.py      	|
|          s3_bucket_tagging.py         	|       s3.py      	|
|           s3_copy_object.py           	|       s3.py      	|
|          s3_delete_objects.py         	|       s3.py      	|
|          s3_file_transform.py         	|       s3.py      	|
|               s3_list.py              	|       s3.py      	|
|          s3_list_prefixes.py          	|       s3.py      	|
|           sagemaker_base.py           	|   sagemaker.py   	|
|         sagemaker_endpoint.py         	|   sagemaker.py   	|
|      sagemaker_endpoint_config.py     	|   sagemaker.py   	|
|           sagemaker_model.py          	|   sagemaker.py   	|
|        sagemaker_processing.py        	|   sagemaker.py   	|
|         sagemaker_training.py         	|   sagemaker.py   	|
|         sagemaker_transform.py        	|   sagemaker.py   	|
|          sagemaker_tuning.py          	|   sagemaker.py   	|
| step_function_get_execution_output.py 	| step_function.py 	|
|    step_function_start_execution.py   	| step_function.py 	|


**Sensors:**

files that don't require changes:

athena.py
batch.py
cloud_formation.py
eks.py
glacier.py
glue.py
glue_crawler.py
redshift.py
sqs.py

files that require changes:

|           Current          	|        New       	|          Note          	|
|:--------------------------:	|:----------------:	|:----------------------:	|
|         dms_task.py        	|      dms.py      	|                        	|
|    ec2_instance_state.py   	|      ec2.py      	|                        	|
|         emr_base.py        	|      emr.py      	|                        	|
|      emr_containers.py     	|      emr.py      	|                        	|
|       emr_job_flow.py      	|      emr.py      	|                        	|
|         emr_step.py        	|      emr.py      	|                        	|
|  glue_catalog_partition.py 	|      glue.py     	| New file already exist 	|
|          s3_key.py         	|       s3.py      	|                        	|
|    s3_keys_unchanged.py    	|       s3.py      	|                        	|
|        s3_prefix.py        	|       s3.py      	|                        	|
|      sagemaker_base.py     	|   sagemaker.py   	|                        	|
|    sagemaker_endpoint.py   	|   sagemaker.py   	|                        	|
|    sagemaker_training.py   	|   sagemaker.py   	|                        	|
|   sagemaker_transform.py   	|   sagemaker.py   	|                        	|
|     sagemaker_tuning.py    	|   sagemaker.py   	|                        	|
| step_function_execution.py 	| step_function.py 	|                        	|


**Action items:**
Create new files and deprecate old ones. This is a backward compatible change.
old file should raise deprecation warnings.

- [x] operators/dms.py @bhavaniravi  https://github.com/apache/airflow/pull/20156
- [x] operators/ec2.py @eladkal  https://github.com/apache/airflow/pull/20157
- [x] operators/emr.py @bhavaniravi https://github.com/apache/airflow/pull/20160
- [x] operators/s3.py @KennyRich https://github.com/apache/airflow/pull/20167
- [x] operators/sagemaker.py @bhavaniravi https://github.com/apache/airflow/pull/20370
- [x] operators/step_function.py @bhavaniravi https://github.com/apache/airflow/pull/20158
- [x] sensors/dms.py @bhavaniravi https://github.com/apache/airflow/pull/20156
- [x] sensors/ec2.py @eladkal  https://github.com/apache/airflow/pull/20157
- [x] sensors/emr.py @bhavaniravi https://github.com/apache/airflow/pull/20160
- ~~[ ] sensors/glue.py  @eladkal On hold as we don't know if this really required. Will be revisited in the future if needed~~
- [x] sensors/s3.py @KennyRich https://github.com/apache/airflow/pull/20167
- [x] sensors/sagemaker.py @bhavaniravi https://github.com/apache/airflow/pull/20370
- [x] sensors/step_function.py @bhavaniravi https://github.com/apache/airflow/pull/20158


FYI @dstandish @ferruzzi @o-nikolas @dbarrundiag 

### Committer

- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.
airflow/providers/amazon/aws/sensors/emr.py
"""
    Asks for the state of the step until it reaches any of the target states.
    If it fails the sensor errors, failing the task.
    With the default target states, sensor waits step to be completed.
    :param job_flow_id: job_flow_id which contains the step check the state of
    :type job_flow_id: str
    :param step_id: step to check the state of
    :type step_id: str
    :param target_states: the target states, sensor waits until
        step reaches any of these states
    :type target_states: list[str]
    :param failed_states: the failure states, sensor fails when
        step reaches any of these states
    :type failed_states: list[str]
    """
    template_fields = ['job_flow_id', 'step_id', 'target_states', 'failed_states']
    template_ext = ()
    def __init__(