We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Unable to specify Python version for AwsGlueJobOperator
### Apache Airflow Provider(s)

amazon

### Versions of Apache Airflow Providers

_No response_

### Apache Airflow version

2.0.2

### Operating System

Amazon Linux

### Deployment

MWAA

### Deployment details

_No response_

### What happened

When a new Glue job is created using the AwsGlueJobOperator, the job is defaulting to Python2. Setting the version in create_job_kwargs fails with key error.

### What you expected to happen

Expected the Glue job to be created with a Python3 runtime. create_job_kwargs are passed to the boto3 glue client create_job method which includes a "Command" parameter that is a dictionary containing the Python version.



### How to reproduce

Create a dag with an AwsGlueJobOperator and pass a "Command" parameter in the create_job_kwargs argument.

```
    create_glue_job_args = {
        "Command": {
            "Name": "abalone-preprocess",
            "ScriptLocation": f"s3://{output_bucket}/code/preprocess.py",
            "PythonVersion": "3"
        }
    }
    glue_etl = AwsGlueJobOperator(  
        task_id="glue_etl",  
        s3_bucket=output_bucket,
        script_args={
                '--S3_INPUT_BUCKET': data_bucket,
                '--S3_INPUT_KEY_PREFIX': 'input/raw',
                '--S3_UPLOADS_KEY_PREFIX': 'input/uploads',
                '--S3_OUTPUT_BUCKET': output_bucket,
                '--S3_OUTPUT_KEY_PREFIX': str(determine_dataset_id.output) +'/input/data' 
            },
        iam_role_name="MLOps",  
        retry_limit=2,
        concurrent_run_limit=3,
        create_job_kwargs=create_glue_job_args,
        dag=dag) 
```

```
[2022-01-04 16:43:42,053] {{logging_mixin.py:104}} INFO - [2022-01-04 16:43:42,053] {{glue.py:190}} ERROR - Failed to create aws glue job, error: 'Command'
[2022-01-04 16:43:42,081] {{logging_mixin.py:104}} INFO - [2022-01-04 16:43:42,081] {{glue.py:112}} ERROR - Failed to run aws glue job, error: 'Command'
[2022-01-04 16:43:42,101] {{taskinstance.py:1482}} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/airflow/providers/amazon/aws/hooks/glue.py", line 166, in get_or_create_glue_job
    get_job_response = glue_client.get_job(JobName=self.job_name)
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 357, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 676, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.errorfactory.EntityNotFoundException: An error occurred (EntityNotFoundException) when calling the GetJob operation: Job with name: abalone-preprocess not found.

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1138, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1311, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1341, in _execute_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/python3.7/site-packages/airflow/providers/amazon/aws/operators/glue.py", line 121, in execute
    glue_job_run = glue_job.initialize_job(self.script_args)
  File "/usr/local/lib/python3.7/site-packages/airflow/providers/amazon/aws/hooks/glue.py", line 108, in initialize_job
    job_name = self.get_or_create_glue_job()
  File "/usr/local/lib/python3.7/site-packages/airflow/providers/amazon/aws/hooks/glue.py", line 186, in get_or_create_glue_job
    **self.create_job_kwargs,
KeyError: 'Command'
```

### Anything else

When a new job is being created.

### Are you willing to submit 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)


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/providers/amazon/aws/hooks/glue.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 
19 import time
20 import warnings
21 from typing import Dict, List, Optional
22 
23 from airflow.exceptions import AirflowException
24 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
25 
26 
27 class GlueJobHook(AwsBaseHook):
28     """
29     Interact with AWS Glue - create job, trigger, crawler
30 
31     :param s3_bucket: S3 bucket where logs and local etl script will be uploaded
32     :param job_name: unique job name per AWS account
33     :param desc: job description
34     :param concurrent_run_limit: The maximum number of concurrent runs allowed for a job
35     :param script_location: path to etl script on s3
36     :param retry_limit: Maximum number of times to retry this job if it fails
37     :param num_of_dpus: Number of AWS Glue DPUs to allocate to this Job
38     :param region_name: aws region name (example: us-east-1)
39     :param iam_role_name: AWS IAM Role for Glue Job Execution
40     :param create_job_kwargs: Extra arguments for Glue Job Creation
41     """
42 
43     JOB_POLL_INTERVAL = 6  # polls job status after every JOB_POLL_INTERVAL seconds
44 
45     def __init__(
46         self,
47         s3_bucket: Optional[str] = None,
48         job_name: Optional[str] = None,
49         desc: Optional[str] = None,
50         concurrent_run_limit: int = 1,
51         script_location: Optional[str] = None,
52         retry_limit: int = 0,
53         num_of_dpus: Optional[int] = None,
54         iam_role_name: Optional[str] = None,
55         create_job_kwargs: Optional[dict] = None,
56         *args,
57         **kwargs,
58     ):
59         self.job_name = job_name
60         self.desc = desc
61         self.concurrent_run_limit = concurrent_run_limit
62         self.script_location = script_location
63         self.retry_limit = retry_limit
64         self.s3_bucket = s3_bucket
65         self.role_name = iam_role_name
66         self.s3_glue_logs = 'logs/glue-logs/'
67         self.create_job_kwargs = create_job_kwargs or {}
68 
69         worker_type_exists = "WorkerType" in self.create_job_kwargs
70         num_workers_exists = "NumberOfWorkers" in self.create_job_kwargs
71 
72         if worker_type_exists and num_workers_exists:
73             if num_of_dpus is not None:
74                 raise ValueError("Cannot specify num_of_dpus with custom WorkerType")
75         elif not worker_type_exists and num_workers_exists:
76             raise ValueError("Need to specify custom WorkerType when specifying NumberOfWorkers")
77         elif worker_type_exists and not num_workers_exists:
78             raise ValueError("Need to specify NumberOfWorkers when specifying custom WorkerType")
79         elif num_of_dpus is None:
80             self.num_of_dpus = 10
81         else:
82             self.num_of_dpus = num_of_dpus
83 
84         kwargs['client_type'] = 'glue'
85         super().__init__(*args, **kwargs)
86 
87     def list_jobs(self) -> List:
88         """:return: Lists of Jobs"""
89         conn = self.get_conn()
90         return conn.get_jobs()
91 
92     def get_iam_execution_role(self) -> Dict:
93         """:return: iam role for job execution"""
94         session, endpoint_url = self._get_credentials(region_name=self.region_name)
95         iam_client = session.client('iam', endpoint_url=endpoint_url, config=self.config, verify=self.verify)
96 
97         try:
98             glue_execution_role = iam_client.get_role(RoleName=self.role_name)
99             self.log.info("Iam Role Name: %s", self.role_name)
100             return glue_execution_role
101         except Exception as general_error:
102             self.log.error("Failed to create aws glue job, error: %s", general_error)
103             raise
104 
105     def initialize_job(
106         self,
107         script_arguments: Optional[dict] = None,
108         run_kwargs: Optional[dict] = None,
109     ) -> Dict[str, str]:
110         """
111         Initializes connection with AWS Glue
112         to run job
113         :return:
114         """
115         glue_client = self.get_conn()
116         script_arguments = script_arguments or {}
117         run_kwargs = run_kwargs or {}
118 
119         try:
120             job_name = self.get_or_create_glue_job()
121             return glue_client.start_job_run(JobName=job_name, Arguments=script_arguments, **run_kwargs)
122 
123         except Exception as general_error:
124             self.log.error("Failed to run aws glue job, error: %s", general_error)
125             raise
126 
127     def get_job_state(self, job_name: str, run_id: str) -> str:
128         """
129         Get state of the Glue job. The job state can be
130         running, finished, failed, stopped or timeout.
131         :param job_name: unique job name per AWS account
132         :param run_id: The job-run ID of the predecessor job run
133         :return: State of the Glue job
134         """
135         glue_client = self.get_conn()
136         job_run = glue_client.get_job_run(JobName=job_name, RunId=run_id, PredecessorsIncluded=True)
137         return job_run['JobRun']['JobRunState']
138 
139     def job_completion(self, job_name: str, run_id: str) -> Dict[str, str]:
140         """
141         Waits until Glue job with job_name completes or
142         fails and return final state if finished.
143         Raises AirflowException when the job failed
144         :param job_name: unique job name per AWS account
145         :param run_id: The job-run ID of the predecessor job run
146         :return: Dict of JobRunState and JobRunId
147         """
148         failed_states = ['FAILED', 'TIMEOUT']
149         finished_states = ['SUCCEEDED', 'STOPPED']
150 
151         while True:
152             job_run_state = self.get_job_state(job_name, run_id)
153             if job_run_state in finished_states:
154                 self.log.info("Exiting Job %s Run State: %s", run_id, job_run_state)
155                 return {'JobRunState': job_run_state, 'JobRunId': run_id}
156             if job_run_state in failed_states:
157                 job_error_message = f"Exiting Job {run_id} Run State: {job_run_state}"
158                 self.log.info(job_error_message)
159                 raise AirflowException(job_error_message)
160             else:
161                 self.log.info(
162                     "Polling for AWS Glue Job %s current run state with status %s", job_name, job_run_state
163                 )
164                 time.sleep(self.JOB_POLL_INTERVAL)
165 
166     def get_or_create_glue_job(self) -> str:
167         """
168         Creates(or just returns) and returns the Job name
169         :return:Name of the Job
170         """
171         glue_client = self.get_conn()
172         try:
173             get_job_response = glue_client.get_job(JobName=self.job_name)
174             self.log.info("Job Already exist. Returning Name of the job")
175             return get_job_response['Job']['Name']
176 
177         except glue_client.exceptions.EntityNotFoundException:
178             self.log.info("Job doesn't exist. Now creating and running AWS Glue Job")
179             if self.s3_bucket is None:
180                 raise AirflowException('Could not initialize glue job, error: Specify Parameter `s3_bucket`')
181             s3_log_path = f's3://{self.s3_bucket}/{self.s3_glue_logs}{self.job_name}'
182             execution_role = self.get_iam_execution_role()
183             try:
184                 if "WorkerType" in self.create_job_kwargs and "NumberOfWorkers" in self.create_job_kwargs:
185                     create_job_response = glue_client.create_job(
186                         Name=self.job_name,
187                         Description=self.desc,
188                         LogUri=s3_log_path,
189                         Role=execution_role['Role']['Arn'],
190                         ExecutionProperty={"MaxConcurrentRuns": self.concurrent_run_limit},
191                         Command={"Name": "glueetl", "ScriptLocation": self.script_location},
192                         MaxRetries=self.retry_limit,
193                         **self.create_job_kwargs,
194                     )
195                 else:
196                     create_job_response = glue_client.create_job(
197                         Name=self.job_name,
198                         Description=self.desc,
199                         LogUri=s3_log_path,
200                         Role=execution_role['Role']['Arn'],
201                         ExecutionProperty={"MaxConcurrentRuns": self.concurrent_run_limit},
202                         Command={"Name": "glueetl", "ScriptLocation": self.script_location},
203                         MaxRetries=self.retry_limit,
204                         MaxCapacity=self.num_of_dpus,
205                         **self.create_job_kwargs,
206                     )
207                 return create_job_response['Name']
208             except Exception as general_error:
209                 self.log.error("Failed to create aws glue job, error: %s", general_error)
210                 raise
211 
212 
213 class AwsGlueJobHook(GlueJobHook):
214     """
215     This hook is deprecated.
216     Please use :class:`airflow.providers.amazon.aws.hooks.glue.GlueJobHook`.
217     """
218 
219     def __init__(self, *args, **kwargs):
220         warnings.warn(
221             "This hook is deprecated. "
222             "Please use :class:`airflow.providers.amazon.aws.hooks.glue.GlueJobHook`.",
223             DeprecationWarning,
224             stacklevel=2,
225         )
226         super().__init__(*args, **kwargs)
227
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/airflow/providers/amazon/aws/hooks/glue.py b/airflow/providers/amazon/aws/hooks/glue.py
--- a/airflow/providers/amazon/aws/hooks/glue.py
+++ b/airflow/providers/amazon/aws/hooks/glue.py
@@ -181,6 +181,12 @@
             s3_log_path = f's3://{self.s3_bucket}/{self.s3_glue_logs}{self.job_name}'
             execution_role = self.get_iam_execution_role()
             try:
+                default_command = {
+                    "Name": "glueetl",
+                    "ScriptLocation": self.script_location,
+                }
+                command = self.create_job_kwargs.get("Command", default_command)
+
                 if "WorkerType" in self.create_job_kwargs and "NumberOfWorkers" in self.create_job_kwargs:
                     create_job_response = glue_client.create_job(
                         Name=self.job_name,
@@ -188,7 +194,7 @@
                         LogUri=s3_log_path,
                         Role=execution_role['Role']['Arn'],
                         ExecutionProperty={"MaxConcurrentRuns": self.concurrent_run_limit},
-                        Command={"Name": "glueetl", "ScriptLocation": self.script_location},
+                        Command=command,
                         MaxRetries=self.retry_limit,
                         **self.create_job_kwargs,
                     )
@@ -199,7 +205,7 @@
                         LogUri=s3_log_path,
                         Role=execution_role['Role']['Arn'],
                         ExecutionProperty={"MaxConcurrentRuns": self.concurrent_run_limit},
-                        Command={"Name": "glueetl", "ScriptLocation": self.script_location},
+                        Command=command,
                         MaxRetries=self.retry_limit,
                         MaxCapacity=self.num_of_dpus,
                         **self.create_job_kwargs,

{"golden_diff": "diff --git a/airflow/providers/amazon/aws/hooks/glue.py b/airflow/providers/amazon/aws/hooks/glue.py\n--- a/airflow/providers/amazon/aws/hooks/glue.py\n+++ b/airflow/providers/amazon/aws/hooks/glue.py\n@@ -181,6 +181,12 @@\n             s3_log_path = f's3://{self.s3_bucket}/{self.s3_glue_logs}{self.job_name}'\n             execution_role = self.get_iam_execution_role()\n             try:\n+                default_command = {\n+                    \"Name\": \"glueetl\",\n+                    \"ScriptLocation\": self.script_location,\n+                }\n+                command = self.create_job_kwargs.get(\"Command\", default_command)\n+\n                 if \"WorkerType\" in self.create_job_kwargs and \"NumberOfWorkers\" in self.create_job_kwargs:\n                     create_job_response = glue_client.create_job(\n                         Name=self.job_name,\n@@ -188,7 +194,7 @@\n                         LogUri=s3_log_path,\n                         Role=execution_role['Role']['Arn'],\n                         ExecutionProperty={\"MaxConcurrentRuns\": self.concurrent_run_limit},\n-                        Command={\"Name\": \"glueetl\", \"ScriptLocation\": self.script_location},\n+                        Command=command,\n                         MaxRetries=self.retry_limit,\n                         **self.create_job_kwargs,\n                     )\n@@ -199,7 +205,7 @@\n                         LogUri=s3_log_path,\n                         Role=execution_role['Role']['Arn'],\n                         ExecutionProperty={\"MaxConcurrentRuns\": self.concurrent_run_limit},\n-                        Command={\"Name\": \"glueetl\", \"ScriptLocation\": self.script_location},\n+                        Command=command,\n                         MaxRetries=self.retry_limit,\n                         MaxCapacity=self.num_of_dpus,\n                         **self.create_job_kwargs,\n", "issue": "Unable to specify Python version for AwsGlueJobOperator\n### Apache Airflow Provider(s)\n\namazon\n\n### Versions of Apache Airflow Providers\n\n_No response_\n\n### Apache Airflow version\n\n2.0.2\n\n### Operating System\n\nAmazon Linux\n\n### Deployment\n\nMWAA\n\n### Deployment details\n\n_No response_\n\n### What happened\n\nWhen a new Glue job is created using the AwsGlueJobOperator, the job is defaulting to Python2. Setting the version in create_job_kwargs fails with key error.\n\n### What you expected to happen\n\nExpected the Glue job to be created with a Python3 runtime. create_job_kwargs are passed to the boto3 glue client create_job method which includes a \"Command\" parameter that is a dictionary containing the Python version.\r\n\r\n\n\n### How to reproduce\n\nCreate a dag with an AwsGlueJobOperator and pass a \"Command\" parameter in the create_job_kwargs argument.\r\n\r\n```\r\n    create_glue_job_args = {\r\n        \"Command\": {\r\n            \"Name\": \"abalone-preprocess\",\r\n            \"ScriptLocation\": f\"s3://{output_bucket}/code/preprocess.py\",\r\n            \"PythonVersion\": \"3\"\r\n        }\r\n    }\r\n    glue_etl = AwsGlueJobOperator(  \r\n        task_id=\"glue_etl\",  \r\n        s3_bucket=output_bucket,\r\n        script_args={\r\n                '--S3_INPUT_BUCKET': data_bucket,\r\n                '--S3_INPUT_KEY_PREFIX': 'input/raw',\r\n                '--S3_UPLOADS_KEY_PREFIX': 'input/uploads',\r\n                '--S3_OUTPUT_BUCKET': output_bucket,\r\n                '--S3_OUTPUT_KEY_PREFIX': str(determine_dataset_id.output) +'/input/data' \r\n            },\r\n        iam_role_name=\"MLOps\",  \r\n        retry_limit=2,\r\n        concurrent_run_limit=3,\r\n        create_job_kwargs=create_glue_job_args,\r\n        dag=dag) \r\n```\r\n\r\n```\r\n[2022-01-04 16:43:42,053] {{logging_mixin.py:104}} INFO - [2022-01-04 16:43:42,053] {{glue.py:190}} ERROR - Failed to create aws glue job, error: 'Command'\r\n[2022-01-04 16:43:42,081] {{logging_mixin.py:104}} INFO - [2022-01-04 16:43:42,081] {{glue.py:112}} ERROR - Failed to run aws glue job, error: 'Command'\r\n[2022-01-04 16:43:42,101] {{taskinstance.py:1482}} ERROR - Task failed with exception\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.7/site-packages/airflow/providers/amazon/aws/hooks/glue.py\", line 166, in get_or_create_glue_job\r\n    get_job_response = glue_client.get_job(JobName=self.job_name)\r\n  File \"/usr/local/lib/python3.7/site-packages/botocore/client.py\", line 357, in _api_call\r\n    return self._make_api_call(operation_name, kwargs)\r\n  File \"/usr/local/lib/python3.7/site-packages/botocore/client.py\", line 676, in _make_api_call\r\n    raise error_class(parsed_response, operation_name)\r\nbotocore.errorfactory.EntityNotFoundException: An error occurred (EntityNotFoundException) when calling the GetJob operation: Job with name: abalone-preprocess not found.\r\n\r\nDuring handling of the above exception, another exception occurred:\r\n\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py\", line 1138, in _run_raw_task\r\n    self._prepare_and_execute_task_with_callbacks(context, task)\r\n  File \"/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py\", line 1311, in _prepare_and_execute_task_with_callbacks\r\n    result = self._execute_task(context, task_copy)\r\n  File \"/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py\", line 1341, in _execute_task\r\n    result = task_copy.execute(context=context)\r\n  File \"/usr/local/lib/python3.7/site-packages/airflow/providers/amazon/aws/operators/glue.py\", line 121, in execute\r\n    glue_job_run = glue_job.initialize_job(self.script_args)\r\n  File \"/usr/local/lib/python3.7/site-packages/airflow/providers/amazon/aws/hooks/glue.py\", line 108, in initialize_job\r\n    job_name = self.get_or_create_glue_job()\r\n  File \"/usr/local/lib/python3.7/site-packages/airflow/providers/amazon/aws/hooks/glue.py\", line 186, in get_or_create_glue_job\r\n    **self.create_job_kwargs,\r\nKeyError: 'Command'\r\n```\n\n### Anything else\n\nWhen a new job is being created.\n\n### Are you willing to submit PR?\n\n- [ ] Yes I am willing to submit a PR!\n\n### Code of Conduct\n\n- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)\n\n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nimport time\nimport warnings\nfrom typing import Dict, List, Optional\n\nfrom airflow.exceptions import AirflowException\nfrom airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook\n\n\nclass GlueJobHook(AwsBaseHook):\n    \"\"\"\n    Interact with AWS Glue - create job, trigger, crawler\n\n    :param s3_bucket: S3 bucket where logs and local etl script will be uploaded\n    :param job_name: unique job name per AWS account\n    :param desc: job description\n    :param concurrent_run_limit: The maximum number of concurrent runs allowed for a job\n    :param script_location: path to etl script on s3\n    :param retry_limit: Maximum number of times to retry this job if it fails\n    :param num_of_dpus: Number of AWS Glue DPUs to allocate to this Job\n    :param region_name: aws region name (example: us-east-1)\n    :param iam_role_name: AWS IAM Role for Glue Job Execution\n    :param create_job_kwargs: Extra arguments for Glue Job Creation\n    \"\"\"\n\n    JOB_POLL_INTERVAL = 6  # polls job status after every JOB_POLL_INTERVAL seconds\n\n    def __init__(\n        self,\n        s3_bucket: Optional[str] = None,\n        job_name: Optional[str] = None,\n        desc: Optional[str] = None,\n        concurrent_run_limit: int = 1,\n        script_location: Optional[str] = None,\n        retry_limit: int = 0,\n        num_of_dpus: Optional[int] = None,\n        iam_role_name: Optional[str] = None,\n        create_job_kwargs: Optional[dict] = None,\n        *args,\n        **kwargs,\n    ):\n        self.job_name = job_name\n        self.desc = desc\n        self.concurrent_run_limit = concurrent_run_limit\n        self.script_location = script_location\n        self.retry_limit = retry_limit\n        self.s3_bucket = s3_bucket\n        self.role_name = iam_role_name\n        self.s3_glue_logs = 'logs/glue-logs/'\n        self.create_job_kwargs = create_job_kwargs or {}\n\n        worker_type_exists = \"WorkerType\" in self.create_job_kwargs\n        num_workers_exists = \"NumberOfWorkers\" in self.create_job_kwargs\n\n        if worker_type_exists and num_workers_exists:\n            if num_of_dpus is not None:\n                raise ValueError(\"Cannot specify num_of_dpus with custom WorkerType\")\n        elif not worker_type_exists and num_workers_exists:\n            raise ValueError(\"Need to specify custom WorkerType when specifying NumberOfWorkers\")\n        elif worker_type_exists and not num_workers_exists:\n            raise ValueError(\"Need to specify NumberOfWorkers when specifying custom WorkerType\")\n        elif num_of_dpus is None:\n            self.num_of_dpus = 10\n        else:\n            self.num_of_dpus = num_of_dpus\n\n        kwargs['client_type'] = 'glue'\n        super().__init__(*args, **kwargs)\n\n    def list_jobs(self) -> List:\n        \"\"\":return: Lists of Jobs\"\"\"\n        conn = self.get_conn()\n        return conn.get_jobs()\n\n    def get_iam_execution_role(self) -> Dict:\n        \"\"\":return: iam role for job execution\"\"\"\n        session, endpoint_url = self._get_credentials(region_name=self.region_name)\n        iam_client = session.client('iam', endpoint_url=endpoint_url, config=self.config, verify=self.verify)\n\n        try:\n            glue_execution_role = iam_client.get_role(RoleName=self.role_name)\n            self.log.info(\"Iam Role Name: %s\", self.role_name)\n            return glue_execution_role\n        except Exception as general_error:\n            self.log.error(\"Failed to create aws glue job, error: %s\", general_error)\n            raise\n\n    def initialize_job(\n        self,\n        script_arguments: Optional[dict] = None,\n        run_kwargs: Optional[dict] = None,\n    ) -> Dict[str, str]:\n        \"\"\"\n        Initializes connection with AWS Glue\n        to run job\n        :return:\n        \"\"\"\n        glue_client = self.get_conn()\n        script_arguments = script_arguments or {}\n        run_kwargs = run_kwargs or {}\n\n        try:\n            job_name = self.get_or_create_glue_job()\n            return glue_client.start_job_run(JobName=job_name, Arguments=script_arguments, **run_kwargs)\n\n        except Exception as general_error:\n            self.log.error(\"Failed to run aws glue job, error: %s\", general_error)\n            raise\n\n    def get_job_state(self, job_name: str, run_id: str) -> str:\n        \"\"\"\n        Get state of the Glue job. The job state can be\n        running, finished, failed, stopped or timeout.\n        :param job_name: unique job name per AWS account\n        :param run_id: The job-run ID of the predecessor job run\n        :return: State of the Glue job\n        \"\"\"\n        glue_client = self.get_conn()\n        job_run = glue_client.get_job_run(JobName=job_name, RunId=run_id, PredecessorsIncluded=True)\n        return job_run['JobRun']['JobRunState']\n\n    def job_completion(self, job_name: str, run_id: str) -> Dict[str, str]:\n        \"\"\"\n        Waits until Glue job with job_name completes or\n        fails and return final state if finished.\n        Raises AirflowException when the job failed\n        :param job_name: unique job name per AWS account\n        :param run_id: The job-run ID of the predecessor job run\n        :return: Dict of JobRunState and JobRunId\n        \"\"\"\n        failed_states = ['FAILED', 'TIMEOUT']\n        finished_states = ['SUCCEEDED', 'STOPPED']\n\n        while True:\n            job_run_state = self.get_job_state(job_name, run_id)\n            if job_run_state in finished_states:\n                self.log.info(\"Exiting Job %s Run State: %s\", run_id, job_run_state)\n                return {'JobRunState': job_run_state, 'JobRunId': run_id}\n            if job_run_state in failed_states:\n                job_error_message = f\"Exiting Job {run_id} Run State: {job_run_state}\"\n                self.log.info(job_error_message)\n                raise AirflowException(job_error_message)\n            else:\n                self.log.info(\n                    \"Polling for AWS Glue Job %s current run state with status %s\", job_name, job_run_state\n                )\n                time.sleep(self.JOB_POLL_INTERVAL)\n\n    def get_or_create_glue_job(self) -> str:\n        \"\"\"\n        Creates(or just returns) and returns the Job name\n        :return:Name of the Job\n        \"\"\"\n        glue_client = self.get_conn()\n        try:\n            get_job_response = glue_client.get_job(JobName=self.job_name)\n            self.log.info(\"Job Already exist. Returning Name of the job\")\n            return get_job_response['Job']['Name']\n\n        except glue_client.exceptions.EntityNotFoundException:\n            self.log.info(\"Job doesn't exist. Now creating and running AWS Glue Job\")\n            if self.s3_bucket is None:\n                raise AirflowException('Could not initialize glue job, error: Specify Parameter `s3_bucket`')\n            s3_log_path = f's3://{self.s3_bucket}/{self.s3_glue_logs}{self.job_name}'\n            execution_role = self.get_iam_execution_role()\n            try:\n                if \"WorkerType\" in self.create_job_kwargs and \"NumberOfWorkers\" in self.create_job_kwargs:\n                    create_job_response = glue_client.create_job(\n                        Name=self.job_name,\n                        Description=self.desc,\n                        LogUri=s3_log_path,\n                        Role=execution_role['Role']['Arn'],\n                        ExecutionProperty={\"MaxConcurrentRuns\": self.concurrent_run_limit},\n                        Command={\"Name\": \"glueetl\", \"ScriptLocation\": self.script_location},\n                        MaxRetries=self.retry_limit,\n                        **self.create_job_kwargs,\n                    )\n                else:\n                    create_job_response = glue_client.create_job(\n                        Name=self.job_name,\n                        Description=self.desc,\n                        LogUri=s3_log_path,\n                        Role=execution_role['Role']['Arn'],\n                        ExecutionProperty={\"MaxConcurrentRuns\": self.concurrent_run_limit},\n                        Command={\"Name\": \"glueetl\", \"ScriptLocation\": self.script_location},\n                        MaxRetries=self.retry_limit,\n                        MaxCapacity=self.num_of_dpus,\n                        **self.create_job_kwargs,\n                    )\n                return create_job_response['Name']\n            except Exception as general_error:\n                self.log.error(\"Failed to create aws glue job, error: %s\", general_error)\n                raise\n\n\nclass AwsGlueJobHook(GlueJobHook):\n    \"\"\"\n    This hook is deprecated.\n    Please use :class:`airflow.providers.amazon.aws.hooks.glue.GlueJobHook`.\n    \"\"\"\n\n    def __init__(self, *args, **kwargs):\n        warnings.warn(\n            \"This hook is deprecated. \"\n            \"Please use :class:`airflow.providers.amazon.aws.hooks.glue.GlueJobHook`.\",\n            DeprecationWarning,\n            stacklevel=2,\n        )\n        super().__init__(*args, **kwargs)\n", "path": "airflow/providers/amazon/aws/hooks/glue.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nimport time\nimport warnings\nfrom typing import Dict, List, Optional\n\nfrom airflow.exceptions import AirflowException\nfrom airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook\n\n\nclass GlueJobHook(AwsBaseHook):\n    \"\"\"\n    Interact with AWS Glue - create job, trigger, crawler\n\n    :param s3_bucket: S3 bucket where logs and local etl script will be uploaded\n    :param job_name: unique job name per AWS account\n    :param desc: job description\n    :param concurrent_run_limit: The maximum number of concurrent runs allowed for a job\n    :param script_location: path to etl script on s3\n    :param retry_limit: Maximum number of times to retry this job if it fails\n    :param num_of_dpus: Number of AWS Glue DPUs to allocate to this Job\n    :param region_name: aws region name (example: us-east-1)\n    :param iam_role_name: AWS IAM Role for Glue Job Execution\n    :param create_job_kwargs: Extra arguments for Glue Job Creation\n    \"\"\"\n\n    JOB_POLL_INTERVAL = 6  # polls job status after every JOB_POLL_INTERVAL seconds\n\n    def __init__(\n        self,\n        s3_bucket: Optional[str] = None,\n        job_name: Optional[str] = None,\n        desc: Optional[str] = None,\n        concurrent_run_limit: int = 1,\n        script_location: Optional[str] = None,\n        retry_limit: int = 0,\n        num_of_dpus: Optional[int] = None,\n        iam_role_name: Optional[str] = None,\n        create_job_kwargs: Optional[dict] = None,\n        *args,\n        **kwargs,\n    ):\n        self.job_name = job_name\n        self.desc = desc\n        self.concurrent_run_limit = concurrent_run_limit\n        self.script_location = script_location\n        self.retry_limit = retry_limit\n        self.s3_bucket = s3_bucket\n        self.role_name = iam_role_name\n        self.s3_glue_logs = 'logs/glue-logs/'\n        self.create_job_kwargs = create_job_kwargs or {}\n\n        worker_type_exists = \"WorkerType\" in self.create_job_kwargs\n        num_workers_exists = \"NumberOfWorkers\" in self.create_job_kwargs\n\n        if worker_type_exists and num_workers_exists:\n            if num_of_dpus is not None:\n                raise ValueError(\"Cannot specify num_of_dpus with custom WorkerType\")\n        elif not worker_type_exists and num_workers_exists:\n            raise ValueError(\"Need to specify custom WorkerType when specifying NumberOfWorkers\")\n        elif worker_type_exists and not num_workers_exists:\n            raise ValueError(\"Need to specify NumberOfWorkers when specifying custom WorkerType\")\n        elif num_of_dpus is None:\n            self.num_of_dpus = 10\n        else:\n            self.num_of_dpus = num_of_dpus\n\n        kwargs['client_type'] = 'glue'\n        super().__init__(*args, **kwargs)\n\n    def list_jobs(self) -> List:\n        \"\"\":return: Lists of Jobs\"\"\"\n        conn = self.get_conn()\n        return conn.get_jobs()\n\n    def get_iam_execution_role(self) -> Dict:\n        \"\"\":return: iam role for job execution\"\"\"\n        session, endpoint_url = self._get_credentials(region_name=self.region_name)\n        iam_client = session.client('iam', endpoint_url=endpoint_url, config=self.config, verify=self.verify)\n\n        try:\n            glue_execution_role = iam_client.get_role(RoleName=self.role_name)\n            self.log.info(\"Iam Role Name: %s\", self.role_name)\n            return glue_execution_role\n        except Exception as general_error:\n            self.log.error(\"Failed to create aws glue job, error: %s\", general_error)\n            raise\n\n    def initialize_job(\n        self,\n        script_arguments: Optional[dict] = None,\n        run_kwargs: Optional[dict] = None,\n    ) -> Dict[str, str]:\n        \"\"\"\n        Initializes connection with AWS Glue\n        to run job\n        :return:\n        \"\"\"\n        glue_client = self.get_conn()\n        script_arguments = script_arguments or {}\n        run_kwargs = run_kwargs or {}\n\n        try:\n            job_name = self.get_or_create_glue_job()\n            return glue_client.start_job_run(JobName=job_name, Arguments=script_arguments, **run_kwargs)\n\n        except Exception as general_error:\n            self.log.error(\"Failed to run aws glue job, error: %s\", general_error)\n            raise\n\n    def get_job_state(self, job_name: str, run_id: str) -> str:\n        \"\"\"\n        Get state of the Glue job. The job state can be\n        running, finished, failed, stopped or timeout.\n        :param job_name: unique job name per AWS account\n        :param run_id: The job-run ID of the predecessor job run\n        :return: State of the Glue job\n        \"\"\"\n        glue_client = self.get_conn()\n        job_run = glue_client.get_job_run(JobName=job_name, RunId=run_id, PredecessorsIncluded=True)\n        return job_run['JobRun']['JobRunState']\n\n    def job_completion(self, job_name: str, run_id: str) -> Dict[str, str]:\n        \"\"\"\n        Waits until Glue job with job_name completes or\n        fails and return final state if finished.\n        Raises AirflowException when the job failed\n        :param job_name: unique job name per AWS account\n        :param run_id: The job-run ID of the predecessor job run\n        :return: Dict of JobRunState and JobRunId\n        \"\"\"\n        failed_states = ['FAILED', 'TIMEOUT']\n        finished_states = ['SUCCEEDED', 'STOPPED']\n\n        while True:\n            job_run_state = self.get_job_state(job_name, run_id)\n            if job_run_state in finished_states:\n                self.log.info(\"Exiting Job %s Run State: %s\", run_id, job_run_state)\n                return {'JobRunState': job_run_state, 'JobRunId': run_id}\n            if job_run_state in failed_states:\n                job_error_message = f\"Exiting Job {run_id} Run State: {job_run_state}\"\n                self.log.info(job_error_message)\n                raise AirflowException(job_error_message)\n            else:\n                self.log.info(\n                    \"Polling for AWS Glue Job %s current run state with status %s\", job_name, job_run_state\n                )\n                time.sleep(self.JOB_POLL_INTERVAL)\n\n    def get_or_create_glue_job(self) -> str:\n        \"\"\"\n        Creates(or just returns) and returns the Job name\n        :return:Name of the Job\n        \"\"\"\n        glue_client = self.get_conn()\n        try:\n            get_job_response = glue_client.get_job(JobName=self.job_name)\n            self.log.info(\"Job Already exist. Returning Name of the job\")\n            return get_job_response['Job']['Name']\n\n        except glue_client.exceptions.EntityNotFoundException:\n            self.log.info(\"Job doesn't exist. Now creating and running AWS Glue Job\")\n            if self.s3_bucket is None:\n                raise AirflowException('Could not initialize glue job, error: Specify Parameter `s3_bucket`')\n            s3_log_path = f's3://{self.s3_bucket}/{self.s3_glue_logs}{self.job_name}'\n            execution_role = self.get_iam_execution_role()\n            try:\n                default_command = {\n                    \"Name\": \"glueetl\",\n                    \"ScriptLocation\": self.script_location,\n                }\n                command = self.create_job_kwargs.get(\"Command\", default_command)\n\n                if \"WorkerType\" in self.create_job_kwargs and \"NumberOfWorkers\" in self.create_job_kwargs:\n                    create_job_response = glue_client.create_job(\n                        Name=self.job_name,\n                        Description=self.desc,\n                        LogUri=s3_log_path,\n                        Role=execution_role['Role']['Arn'],\n                        ExecutionProperty={\"MaxConcurrentRuns\": self.concurrent_run_limit},\n                        Command=command,\n                        MaxRetries=self.retry_limit,\n                        **self.create_job_kwargs,\n                    )\n                else:\n                    create_job_response = glue_client.create_job(\n                        Name=self.job_name,\n                        Description=self.desc,\n                        LogUri=s3_log_path,\n                        Role=execution_role['Role']['Arn'],\n                        ExecutionProperty={\"MaxConcurrentRuns\": self.concurrent_run_limit},\n                        Command=command,\n                        MaxRetries=self.retry_limit,\n                        MaxCapacity=self.num_of_dpus,\n                        **self.create_job_kwargs,\n                    )\n                return create_job_response['Name']\n            except Exception as general_error:\n                self.log.error(\"Failed to create aws glue job, error: %s\", general_error)\n                raise\n\n\nclass AwsGlueJobHook(GlueJobHook):\n    \"\"\"\n    This hook is deprecated.\n    Please use :class:`airflow.providers.amazon.aws.hooks.glue.GlueJobHook`.\n    \"\"\"\n\n    def __init__(self, *args, **kwargs):\n        warnings.warn(\n            \"This hook is deprecated. \"\n            \"Please use :class:`airflow.providers.amazon.aws.hooks.glue.GlueJobHook`.\",\n            DeprecationWarning,\n            stacklevel=2,\n        )\n        super().__init__(*args, **kwargs)\n", "path": "airflow/providers/amazon/aws/hooks/glue.py"}]}