From 23dff6a5678a9a47d5999eb4a6d32269adccf0de Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Wed, 2 Mar 2022 12:22:34 -0800
Subject: [PATCH] EMR on EKS Sample DAG and Docs Update

---
 .../aws/example_dags/example_emr_eks_job.py   | 20 ++++------
 airflow/providers/amazon/aws/hooks/emr.py     |  2 +-
 .../operators/emr_eks.rst                     | 38 +++++++++++++------
 3 files changed, 35 insertions(+), 25 deletions(-)

diff --git a/airflow/providers/amazon/aws/example_dags/example_emr_eks_job.py b/airflow/providers/amazon/aws/example_dags/example_emr_eks_job.py
index 11c1c5b5f6ffa..e8932630d9d45 100644
--- a/airflow/providers/amazon/aws/example_dags/example_emr_eks_job.py
+++ b/airflow/providers/amazon/aws/example_dags/example_emr_eks_job.py
@@ -14,20 +14,15 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-"""
-This is an example dag for an Amazon EMR on EKS Spark job.
-"""
+
 import os
-from datetime import datetime, timedelta
+from datetime import datetime
 
 from airflow import DAG
 from airflow.providers.amazon.aws.operators.emr import EmrContainerOperator
 
-# [START howto_operator_emr_eks_env_variables]
 VIRTUAL_CLUSTER_ID = os.getenv("VIRTUAL_CLUSTER_ID", "test-cluster")
 JOB_ROLE_ARN = os.getenv("JOB_ROLE_ARN", "arn:aws:iam::012345678912:role/emr_eks_default_role")
-# [END howto_operator_emr_eks_env_variables]
-
 
 # [START howto_operator_emr_eks_config]
 JOB_DRIVER_ARG = {
@@ -56,19 +51,18 @@
 # [END howto_operator_emr_eks_config]
 
 with DAG(
-    dag_id='emr_eks_pi_job',
-    dagrun_timeout=timedelta(hours=2),
+    dag_id='example_emr_eks_job',
+    schedule_interval=None,
     start_date=datetime(2021, 1, 1),
-    schedule_interval="@once",
+    tags=['example'],
     catchup=False,
-    tags=["emr_containers", "example"],
 ) as dag:
 
     # An example of how to get the cluster id and arn from an Airflow connection
     # VIRTUAL_CLUSTER_ID = '{{ conn.emr_eks.extra_dejson["virtual_cluster_id"] }}'
     # JOB_ROLE_ARN = '{{ conn.emr_eks.extra_dejson["job_role_arn"] }}'
 
-    # [START howto_operator_emr_eks_jobrun]
+    # [START howto_operator_emr_eks_job]
     job_starter = EmrContainerOperator(
         task_id="start_job",
         virtual_cluster_id=VIRTUAL_CLUSTER_ID,
@@ -78,4 +72,4 @@
         configuration_overrides=CONFIGURATION_OVERRIDES_ARG,
         name="pi.py",
     )
-    # [END howto_operator_emr_eks_jobrun]
+    # [END howto_operator_emr_eks_job]
diff --git a/airflow/providers/amazon/aws/hooks/emr.py b/airflow/providers/amazon/aws/hooks/emr.py
index 271821cb614b1..9a2c13ca9d684 100644
--- a/airflow/providers/amazon/aws/hooks/emr.py
+++ b/airflow/providers/amazon/aws/hooks/emr.py
@@ -135,7 +135,7 @@ def submit_job(
         client_request_token: Optional[str] = None,
     ) -> str:
         """
-        Submit a job to the EMR Containers API and and return the job ID.
+        Submit a job to the EMR Containers API and return the job ID.
         A job run is a unit of work, such as a Spark jar, PySpark script,
         or SparkSQL query, that you submit to Amazon EMR on EKS.
         See: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/emr-containers.html#EMRContainers.Client.start_job_run  # noqa: E501
diff --git a/docs/apache-airflow-providers-amazon/operators/emr_eks.rst b/docs/apache-airflow-providers-amazon/operators/emr_eks.rst
index d59c86693d1b0..397bc1cb4d98c 100644
--- a/docs/apache-airflow-providers-amazon/operators/emr_eks.rst
+++ b/docs/apache-airflow-providers-amazon/operators/emr_eks.rst
@@ -21,16 +21,21 @@
 Amazon EMR on EKS Operators
 ===========================
 
-`Amazon EMR on EKS <https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/emr-eks.html>`__  provides a deployment option for Amazon EMR that allows you to run open-source big data frameworks on Amazon Elastic Kubernetes Service (Amazon EKS).
+`Amazon EMR on EKS <https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/emr-eks.html>`__
+provides a deployment option for Amazon EMR that allows you to run open-source big data frameworks on
+Amazon EKS.
 
-Airflow provides the :class:`~airflow.providers.amazon.aws.operators.emr.EmrContainerOperator` to submit Spark jobs to your EMR on EKS virtual cluster.
+Airflow provides the :class:`~airflow.providers.amazon.aws.operators.emr.EmrContainerOperator`
+to submit Apache Spark jobs to your EMR on EKS virtual cluster.
 
 Prerequisite Tasks
 ------------------
 
 .. include:: _partials/prerequisite_tasks.rst
 
-This example assumes that you already have an EMR on EKS virtual cluster configured. See the `EMR on EKS Getting Started guide <https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/getting-started.html>`__ for more information.
+This example assumes that you already have an EMR on EKS virtual cluster configured. See the
+`EMR on EKS Getting Started guide <https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/getting-started.html>`__
+for more information.
 
 
 Run a Spark job on EMR on EKS
@@ -39,16 +44,23 @@ Run a Spark job on EMR on EKS
 Purpose
 """""""
 
-The ``EMRContainerOperator`` will submit a new job to an EMR on EKS virtual cluster and wait for the job to complete. The example job below calculates the mathematical constant ``Pi``, and monitors the progress with ``EmrContainerSensor``. In a production job, you would usually refer to a Spark script on Amazon S3.
+The ``EMRContainerOperator`` will submit a new job to an EMR on EKS virtual cluster and wait for
+the job to complete. The example job below calculates the mathematical constant ``Pi``. In a
+production job, you would usually refer to a Spark script on Amazon Simple Storage Service (S3).
 
 Job configuration
 """""""""""""""""
 
-To create a job for EMR on EKS, you need to specify your virtual cluster ID, the release of EMR you want to use, your IAM execution role, and Spark submit parameters.
+To create a job for EMR on EKS, you need to specify your virtual cluster ID, the release of EMR you
+want to use, your IAM execution role, and Spark submit parameters.
 
-You can also optionally provide configuration overrides such as Spark, Hive, or Log4j properties as well as monitoring configuration that sends Spark logs to S3 or Cloudwatch.
+You can also optionally provide configuration overrides such as Spark, Hive, or Log4j properties as
+well as monitoring configuration that sends Spark logs to S3 or Amazon Cloudwatch.
 
-In the example, we show how to add an ``applicationConfiguration`` to use the AWS Glue data catalog and ``monitoringConfiguration`` to send logs to the ``/aws/emr-eks-spark`` log group in CloudWatch. Refer to the `EMR on EKS guide <https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/emr-eks-jobs-CLI.html#emr-eks-jobs-parameters>`__ for more details on job configuration.
+In the example, we show how to add an ``applicationConfiguration`` to use the AWS Glue data catalog
+and ``monitoringConfiguration`` to send logs to the ``/aws/emr-eks-spark`` log group in CloudWatch.
+Refer to the `EMR on EKS guide <https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/emr-eks-jobs-CLI.html#emr-eks-jobs-parameters>`__
+for more details on job configuration.
 
 .. exampleinclude:: /../../airflow/providers/amazon/aws/example_dags/example_emr_eks_job.py
     :language: python
@@ -56,15 +68,19 @@ In the example, we show how to add an ``applicationConfiguration`` to use the AW
     :end-before: [END howto_operator_emr_eks_config]
 
 
-We pass the ``virtual_cluster_id`` and ``execution_role_arn`` values as operator parameters, but you can store them in a connection or provide them in the DAG. Your AWS region should be defined either in the ``aws_default`` connection as ``{"region_name": "us-east-1"}`` or a custom connection name that gets passed to the operator with the ``aws_conn_id`` parameter.
+We pass the ``virtual_cluster_id`` and ``execution_role_arn`` values as operator parameters, but you
+can store them in a connection or provide them in the DAG. Your AWS region should be defined either
+in the ``aws_default`` connection as ``{"region_name": "us-east-1"}`` or a custom connection name
+that gets passed to the operator with the ``aws_conn_id`` parameter.
 
 .. exampleinclude:: /../../airflow/providers/amazon/aws/example_dags/example_emr_eks_job.py
     :language: python
     :dedent: 4
-    :start-after: [START howto_operator_emr_eks_jobrun]
-    :end-before: [END howto_operator_emr_eks_jobrun]
+    :start-after: [START howto_operator_emr_eks_job]
+    :end-before: [END howto_operator_emr_eks_job]
 
-With the EmrContainerOperator, it will wait until the successful completion of the job or raise an ``AirflowException`` if there is an error. The operator returns the Job ID of the job run.
+With the EmrContainerOperator, it will wait until the successful completion of the job or raise
+an ``AirflowException`` if there is an error. The operator returns the Job ID of the job run.
 
 Reference
 ---------
