### Body

Use `AirflowProviderDeprecationWarning` as warning source in EMR operators and change stacklevel to 2:
https://github.com/apache/airflow/blob/05036e619c0c6dafded1451daac4e07e20aee33f/airflow/providers/amazon/aws/operators/emr.py#L380-L384
https://github.com/apache/airflow/blob/05036e619c0c6dafded1451daac4e07e20aee33f/airflow/providers/amazon/aws/operators/emr.py#L373-L377
https://github.com/apache/airflow/blob/05036e619c0c6dafded1451daac4e07e20aee33f/airflow/providers/amazon/aws/operators/emr.py#L264-L268
https://github.com/apache/airflow/blob/05036e619c0c6dafded1451daac4e07e20aee33f/airflow/providers/amazon/aws/operators/emr.py#L257-L261

### Committer

- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.
"""Create and return an EmrContainerHook."""
        return EmrContainerHook(
            self.aws_conn_id,
            virtual_cluster_id=self.virtual_cluster_id,
        )
    def execute(self, context: Context) -> str | None:
        """Run job on EMR Containers."""
        self.job_id = self.hook.submit_job(
            self.name,
            self.execution_role_arn,
            self.release_label,
            self.job_driver,
            self.configuration_overrides,
            self.client_request_token,
            self.tags,
        )
        if self.deferrable:
            query_status = self.hook.check_query_status(job_id=self.job_id)
            self.check_failure(query_status)
            if query_status in EmrContainerHook.SUCCESS_STATES:
                return self.job_id
            timeout = (
                timedelta(seconds=self.max_polling_attempts * self.poll_interval)
                if self.max_polling_attempts
                else self.execution_timeout
            )
            self.defer(