### Description

Add feature to EmrStepSensor to bring back the spark task url & logs after task execution

### Use case/motivation

After starting an EMR step task using EmrAddStepsOperator we generally have an EmrStepSensor to track the status of the step. The job ID is available for the sensor and is being poked at regular interval. 

```
[2022-04-26, 22:07:43 UTC] {base_aws.py:100} INFO - Retrieving region_name from Connection.extra_config['region_name']
[2022-04-26, 22:07:44 UTC] {emr.py:316} INFO - Poking step s-123ABC123ABC on cluster j-123ABC123ABC
[2022-04-26, 22:07:44 UTC] {emr.py:74} INFO - Job flow currently PENDING
[2022-04-26, 22:08:44 UTC] {emr.py:316} INFO - Poking step s-123ABC123ABC on cluster j-123ABC123ABC
[2022-04-26, 22:08:44 UTC] {emr.py:74} INFO - Job flow currently PENDING
[2022-04-26, 22:09:44 UTC] {emr.py:316} INFO - Poking step s-123ABC123ABC on cluster j-123ABC123ABC
[2022-04-26, 22:09:44 UTC] {emr.py:74} INFO - Job flow currently COMPLETED
[2022-04-26, 22:09:44 UTC] {base.py:251} INFO - Success criteria met. Exiting.
[2022-04-26, 22:09:44 UTC] {taskinstance.py:1288} INFO - Marking task as SUCCESS. dag_id=datapipeline_sample, task_id=calculate_pi_watch_step, execution_date=20220426T220739, start_date=20220426T220743, end_date=20220426T220944
```

After the task is completed the status is displayed. If the user wants to review the logs of the task, it is a multistep process to get hold of the job logs from EMR cluster. 

It will be a great addition to add the log url and possibly relay the logs to Airflow EmrStepSensor post completion of the task. This will be very handy when there are failures of many tasks and will make it a great user experience. 


### Related issues

_No response_

### Are you willing to submit a PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

boto infrastructure to ship a file to s3.
        :param string_data: str to set as content for the key.
        :param key: S3 key that will point to the file
        :param bucket_name: Name of the bucket in which to store the file
        :param replace: A flag to decide whether or not to overwrite the key
            if it already exists
        :param encrypt: If True, the file will be encrypted on the server-side
            by S3 and will be stored in an encrypted form while at rest in S3.
        :param encoding: The string to byte encoding
        :param acl_policy: The string to specify the canned ACL policy for the
            object to be uploaded
        :param compression: Type of compression to use, currently only gzip is supported.
        """
        encoding = encoding or "utf-8"
        bytes_data = string_data.encode(encoding)
        
        available_compressions = ["gzip"]
        if compression is not None and compression not in available_compressions:
            raise NotImplementedError(
                f"Received {compression} compression type. "
                f"String can currently be compressed in {available_compressions} only."
            )
        if compression == "gzip":
            bytes_data = gz.compress(bytes_data)
        file_obj = io.BytesIO(bytes_data)
        self._upload_file_obj(file_obj, key, bucket_name, replace, encrypt, acl_policy)
        file_obj.close()
    @provide_bucket_name
    @unify_bucket_name_and_key
    def load_bytes(