[AIRFLOW-2310]: Add AWS Glue Job Compatibility to Airflow by oelesinsc24 · Pull Request #4068 · apache/airflow

oelesinsc24 · 2018-10-19T05:45:25Z

Make sure you have checked all steps below.

Jira

My PR addresses the following Airflow Jira issues and references them in the PR title. For example, "[AIRFLOW-XXX] My Airflow PR"
- https://issues.apache.org/jira/projects/AIRFLOW/issues/AIRFLOW-2310
- In case you are fixing a typo in the documentation you can prepend your commit with [AIRFLOW-XXX], code changes always need a Jira issue.

Description

Here are some details about my PR, including screenshots of any UI changes:
This PR is a continuation of the existing PR: [AIRFLOW-2310]: Add AWS Glue Job Compatibility to Airflow #3504.
Adds AWS Glue execution to Apache Airflow

Tests

My PR adds the following unit tests OR does not need testing for this extremely good reason:
Added tests.contrib.test_aws_glue_job_hook.py.
However, moto the test class for boto3 does not support AWS Glue Job mock currently

Commits

My commits all reference Jira issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters (not including Jira issue reference)
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"

Documentation

In case of new functionality, my PR adds documentation that describes how to use it.
- When adding new operators/hooks/sensors, the autoclass documentation generation needs to be added.

Code Quality

Passes flake8

…tream issues

…b and not create glue job

…elated errors

oelesinsc24 · 2018-10-19T14:29:40Z

@Fokko, can you please help look at this?

Some Kubernetes tests seem to be failing and they are not related to my commits, precisely, I have not had anything to do with them

Fokko

@oelesinsc24 Thanks for the PR, please check the comments.

oelesinsc24 · 2018-10-29T14:48:35Z

@Fokko , thanks for the review. Will the changes be merged any time soon?

Fokko · 2018-11-05T18:14:02Z

@oelesinsc24 Travis is still not happy:

======================================================================
26) ERROR: test_execute_without_failure (tests.contrib.operators.test_aws_glue_job_operator.TestAwsGlueJobOperator)
----------------------------------------------------------------------
   Traceback (most recent call last):
    .tox/py27-backend_mysql-env_docker/lib/python2.7/site-packages/mock/mock.py line 1305 in patched
      return func(*args, **keywargs)
    tests/contrib/operators/test_aws_glue_job_operator.py line 54 in test_execute_without_failure
      self.glue.execute(None)
    airflow/contrib/operators/aws_glue_job_operator.py line 68 in execute
      region_name=self.region_name
   AttributeError: 'AWSGlueJobOperator' object has no attribute 'region_name'

…by @letianw91 on 08-11-2018

nwagh · 2018-11-15T02:13:02Z

Hi, Any update on this PR. very interesting hook

infa-rasrinivasan · 2018-12-01T07:25:23Z

Is there any further update on when this PR will be actioned on closed ?

oelesinsc24 · 2018-12-01T08:48:31Z

Is there any further update on when this PR will be actioned on closed ?

As far as I know, the PR is approved. Awaiting merge. @Fokko, can you help see to this? Thanks

ashb

Code looks good.

Only question is about the split of the Operator and the sensor: is there ever a case where you want a glue job to "fire-and-forget" or would you the AwsGlueJobOperator to handle the polling internally, without the need for a sensor task?

oelesinsc24 · 2018-12-04T13:53:08Z

@ashb, great question. But I think there may be cases for both. However, I would recommend submitting a glue job and then using the sensor to poll for updates.

ashb · 2018-12-04T14:19:56Z

I would recommend submitting a glue job and then using the sensor to poll for updates.

Could you explain your thinking? If this is what 90% of people will want, then does it make more sense having it be the default and built as a single operator?

shayan90 · 2018-12-04T20:48:56Z

This is great, when are you guys planning to merge this ?
@oelesinsc24 it might be useful to fully leverage Glue API, two usefull arguments are AllocatedCapacity for number of DPUs per Job Run and SecurityConfiguration would allow encryption at rest using KMS.
https://github.com/apache/incubator-airflow/blob/71cb37eaedae50e2abe9a13591e31a56f2e3659e/airflow/contrib/hooks/aws_glue_job_hook.py#L67-L70

oelesinsc24 · 2018-12-12T11:51:05Z

I would recommend submitting a glue job and then using the sensor to poll for updates.

Could you explain your thinking? If this is what 90% of people will want, then does it make more sense having it be the default and built as a single operator?

I guess if 90% of people want this then we can add it. However, in my experience with AWS Glue, it makes sense to submit the job and then poll for it

ashb · 2018-12-12T13:02:12Z

Do you ever do anything else than Submit the job and the immediately start a sensor to poll for it?

Fokko · 2018-12-13T08:17:54Z

Like stated earlier. My preference would be to have the operator itself poll for it. If you think about atomicity, if the job fails, you will need to clear both the sensor and the operator. This makes life more complicated and adds very little value. Also, your dags will be twice as big.

Please fix the flake8 issues: ./tests/contrib/operators/test_aws_glue_job_operator.py:42:9: F841 local variable 'some_script' is assigned to but never used

oelesinsc24 · 2018-12-13T08:59:09Z

Do you ever do anything else than Submit the job and the immediately start a sensor to poll for it?

Now I get the point and I agree as well. I will add the necessary changes. @Fokko , thanks for the clarification as well.

Fokko · 2018-12-13T09:15:06Z

Any time @oelesinsc24. Let me know when you're ready, so we can get this in! 👍

amelio-vazquez-reina · 2018-12-18T04:11:24Z

How would this work? I assume the Glue endpoint has to be setup a priori with the code that it would run?

zingorn · 2019-01-15T11:12:14Z

+                              .format(job_name))
+                time.sleep(6)
+
+    def get_glue_job(self):


this code can only create a job but I want to run exists job by name
so suggest to use

def get_or_create_glue_job(self): try: glue_client = self.get_conn() glue_client.get_job(JobName=self.job_name) except ClientError as client_err: if client_err.response['Error']['Code'] == \ 'EntityNotFoundException': return self._create_glue_job() raise client_err def _create_glue_job(self): glue_client = self.get_conn() if self.s3_bucket is None: raise AirflowException( 'Could not initialize glue job, ' 'error: Specify Parameter `s3_bucket`' ) self.log.info("Now creating and running AWS Glue Job") s3_log_path = "s3://{bucket_name}/{logs_path}{job_name}"\ .format(bucket_name=self.s3_bucket, logs_path=self.S3_GLUE_LOGS, job_name=self.job_name) execution_role = self.get_iam_execution_role() script_location = self._check_script_location() create_job_response = glue_client.create_job( Name=self.job_name, Description=self.desc, LogUri=s3_log_path, Role=execution_role['Role']['RoleName'], ExecutionProperty={"MaxConcurrentRuns": self.concurrent_run_limit}, Command={"Name": "glueetl", "ScriptLocation": script_location}, MaxRetries=self.retry_limit, AllocatedCapacity=self.num_of_dpus ) return create_job_response

and then in initialize_job do

glue_client = self.get_conn() try: self.get_or_create_glue_job() job_run = glue_client.start_job_run( JobName=self.job_name, Arguments=script_arguments ) self.log.info("Run job with id: {}".format(job_run['JobRunId'])) return self.job_completion(self.job_name, job_run['JobRunId']) except Exception as general_error: raise AirflowException( 'Failed to run aws glue job, error: {error}'.format( error=str(general_error) ) )

zingorn · 2019-01-15T11:13:41Z

+            )
+            job_run_state = job_status['JobRun']['JobRunState']
+
+            if job_run_state in {'FAILED', 'STOPPED', 'SUCCEEDED'}:


use constants for the states, then we can you them in tests

zingorn · 2019-01-15T11:15:26Z

+    :param aws_conn_id: aws connection id
+    :type aws_conn_id: str
+    """
+    template_fields = ()


please, add script_args argument to allow templating

template_fields = ('script_args',)

+1 , IMO it may be useful to template the job_name as well.

zingorn · 2019-01-15T11:54:59Z

+                RunId=run_id,
+                PredecessorsIncluded=True
+            )
+            job_run_state = job_status['JobRun']['JobRunState']


collect an error message for logging

error_message = job_status['JobRun']['ErrorMessage'] \ if 'ErrorMessage' in job_status['JobRun'] else None if failed or stopped or completed: self.log.info("Exiting Job {} Run State: {}" .format(run_id, job_run_state)) return {'JobRunState': job_run_state, 'JobRunId': run_id, 'ErrorMessage': error_message}

then in AWSGlueJobOperator

glue_job_run = glue_job.initialize_job(self.script_args) error_message = 'Error message: {}'.format(glue_job_run['ErrorMessage'])\ if glue_job_run['ErrorMessage'] else '' self.log.info('AWS Glue Job: {job_name} status: {job_status}. ' 'Run Id: {run_id}{error_message}' .format(run_id=glue_job_run['JobRunId'], job_name=self.job_name, job_status=glue_job_run['JobRunState'], error_message=error_message) ) if glue_job_run['JobRunState'] != SUCCEEDED: raise AirflowException('AWS Glue job failed with error: {}' .format(glue_job_run['ErrorMessage']))

mik-laj · 2019-01-20T00:14:40Z

+            get_job_response = glue_client.get_job(
+                JobName=self.job_name
+            )['Job']
+            self.logger.info("Found AWS Glue Job: {job_name}".format(job_name=self.job_name))


Suggested change

self.logger.info("Found AWS Glue Job: {job_name}".format(job_name=self.job_name))

self.logger.info("Found AWS Glue Job: %s", self.job_name)

Please avoid formatting string. You should pass variables to logger to avoid creating string objects in memory. When the logger level is too low, they will never be created in memory.

mik-laj · 2019-01-22T00:55:27Z

+                )
+            )
+
+    def job_completion(self, job_name=None, run_id=None):


Could you correct the name of this method to include a verb? GoogleCloudBaseHook use a name
_wait_for_operation_to_complete.

See also https://github.com/apache/airflow/blob/master/airflow/contrib/operators/awsbatch_operator.py#L125

python doesn't have java-interfaces, but that seems be what this requires; is there a general Airflow operator ABC, document or recommendation on all these operator method names?

mik-laj · 2019-02-08T09:16:27Z

Could you complete the integration.rst file?

kurtqq · 2019-03-11T20:19:33Z

I'm looking forward to this PR

kyle-hamlin · 2019-05-07T20:50:46Z

@oelesinsc24 @Fokko @ashb Any update on this PR? Also, I agree that the operator should follow the status of the job, there is really no need for that sensor.

zhongjiajie

My 2 cent, most of them because AIP-3 According to https://issues.apache.org/jira/browse/AIRFLOW-4196

zhongjiajie · 2019-05-08T01:44:25Z

+        self.job_name = job_name
+        self.desc = desc
+        self.aws_conn_id = aws_conn_id
+        super(AwsGlueJobHook, self).__init__(self.aws_conn_id, *args, **kwargs)


Suggested change

super(AwsGlueJobHook, self).__init__(self.aws_conn_id, *args, **kwargs)

super().__init__(self.aws_conn_id, *args, **kwargs)

zhongjiajie · 2019-05-08T01:47:30Z

+                 aws_conn_id='aws_default',
+                 *args, **kwargs
+                 ):
+        super(AWSGlueJobOperator, self).__init__(*args, **kwargs)


Suggested change

super(AWSGlueJobOperator, self).__init__(*args, **kwargs)

super().__init__(*args, **kwargs)

According to https://issues.apache.org/jira/browse/AIRFLOW-4196

zhongjiajie · 2019-05-08T01:47:49Z

+                 aws_conn_id='aws_default',
+                 *args,
+                 **kwargs):
+        super(AwsGlueJobSensor, self).__init__(*args, **kwargs)


Suggested change

super(AwsGlueJobSensor, self).__init__(*args, **kwargs)

super().__init__(*args, **kwargs)

According to https://issues.apache.org/jira/browse/AIRFLOW-4196

zhongjiajie · 2019-05-08T01:48:13Z

 .. autoclass:: airflow.contrib.operators.adls_to_gcs.AdlsToGoogleCloudStorageOperator
 .. autoclass:: airflow.contrib.operators.aws_athena_operator.AWSAthenaOperator
 .. autoclass:: airflow.contrib.operators.awsbatch_operator.AWSBatchOperator
+.. autoclass:: airflow.contrib.operators.aws_glue_job_operator.AWSGlueJobOperator


We don't need code.rst anymore

zhongjiajie · 2019-05-08T01:50:41Z

+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import unicode_literals


Suggested change

from __future__ import unicode_literals

don't need __future__ anymore

According to https://issues.apache.org/jira/browse/AIRFLOW-4196

mrshu · 2019-05-14T17:21:23Z

I'd be very interested in getting this merged -- please do let me know how I can be of help here.

dgokeeffe · 2019-07-22T07:04:42Z

I'm also interested in getting this PR approved -- I can spend some time getting the operator to poll for the status of the job itself.

oelesinsc24 · 2019-08-19T11:49:35Z

I see the merge failed from what is (hopefully?) a small conflict - can we get eyes on this? can I help?

Yes you can help. Apologies, I've been really busy for some time now.

oelesinsc24 · 2019-08-19T11:50:38Z

I'd be very interested in getting this merged -- please do let me know how I can be of help here.

Please do, all hands needed to get this approved and available to all

dazza-codes · 2019-08-29T14:58:22Z

watching
2c on operator and/or sensor: a sensor has an option to be rescheduled for the next poll, so it can release a worker, whereas e.g. the batch operator ties up a worker while it polls for a job status (I recently contributed to that piece of code so it polls less frequently, but it still ties up a worker); ideally, it would be great if an operator that needs to poll for status could be rescheduled in the same way that a sensor can be rescheduled (not sure what that refactor looks like; and not asking for it here in this PR)
- https://github.com/apache/airflow/blob/master/airflow/contrib/operators/awsbatch_operator.py#L125

dazza-codes · 2019-08-29T15:05:00Z

https://github.com/oelesinsc24 has very little activity on github profile
@oelesinsc24 - status update - is it abandoned or will it get updated on your fork?
- all commits are dated in late 2018, nothing in 2019
looks like it needs:
- drop master-merge commits
- do a rebase on master and resolve conflicts
  - some comments about dropping code.rst for some reason
- resolve review comments and suggested changes
  - fix some super calls
  - maybe punt more substantial refactors to follow-up PRs?
- squash all commits into one feature commit
- get the travis CI passing

oelesinsc24 · 2019-08-30T07:47:48Z

https://github.com/oelesinsc24 has very little activity on github profile

@oelesinsc24 - status update - is it abandoned or will it get updated on your fork?

all commits are dated in late 2018, nothing in 2019

looks like it needs:

drop master-merge commits

do a rebase on master and resolve conflicts

some comments about dropping code.rst for some reason

resolve review comments and suggested changes

fix some super calls

maybe punt more substantial refactors to follow-up PRs?

squash all commits into one feature commit

get the travis CI passing

The github profile is private, here's my public: https://github.com/OELEsin. Good you asked about the status. I have not had much time to contribute to it. This is why I asked if anyone was willing to continue from where I stopped

kaxil · 2019-09-04T11:52:16Z

Looks like there is a good amount of demand for this one. If @oelesinsc24 is busy, does anyone want to take this forward, create a new PR?

abdulbasitds · 2019-09-04T11:57:29Z

Hi,I wanted to work on it, I cloned this repo (https://github.com/OElesin/incubator-airflow) and tried the glue sensor/hook/operator and they weren't working properly so i have fixed and have submiited this pull request(#6007)

stale · 2019-11-20T16:52:22Z

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

oelesinsc24 added 2 commits October 18, 2018 16:00

[AIRFLOW-2310]: Add AWS Glue Job Compatibility to Airflow and fix ups…

466c272

…tream issues

[AIRFLOW-2310]: Changed AWS Glue Hook to only run already existing jo…

523d739

…b and not create glue job

oelesinsc24 mentioned this pull request Oct 19, 2018

[AIRFLOW-2310]: Add AWS Glue Job Compatibility to Airflow #3504

Closed

5 tasks done

[AIRFLOW-2310]: Fixed flake8 issues; Failing builds were due to non r…

dafb2fb

…elated errors

Fokko requested changes Oct 20, 2018

View reviewed changes

[AIRFLOW-2310]: Fixed changes based on suggestions from @Fokko

2193e32

Fokko requested changes Oct 23, 2018

View reviewed changes

Comment thread airflow/contrib/hooks/aws_glue_job_hook.py Resolved

Comment thread airflow/contrib/hooks/aws_glue_job_hook.py Resolved

Comment thread airflow/contrib/operators/aws_glue_job_operator.py Resolved

[AIRFLOW-2310]: Fixed changes based on review from @Fokko on 23-10-2018

bd3e5e3

ashb reviewed Oct 23, 2018

View reviewed changes

Comment thread airflow/contrib/hooks/aws_glue_job_hook.py Resolved

ashb reviewed Oct 23, 2018

View reviewed changes

Comment thread airflow/contrib/operators/aws_glue_job_operator.py Resolved

[AIRFLOW-2310]: Fixed docstrings as requested from @ashb on 23-10-2018

4bbb9b7

[AIRFLOW-2310]: Fixed test bugs as requested from @Fokko on 05-11-2018

24e6e91

lwbayes reviewed Nov 8, 2018

View reviewed changes

Comment thread airflow/contrib/hooks/aws_glue_job_hook.py Resolved

oelesinsc24 added 2 commits November 8, 2018 16:18

[AIRFLOW-2310]: Fixed while loop to poll glue job status as reviewed …

9d2f500

…by @letianw91 on 08-11-2018

[AIRFLOW-2310]: Resolved failing tests due to previous refactoring

1b1ce9a

Fokko approved these changes Nov 13, 2018

View reviewed changes

Merge branch 'master' into AIRFLOW-2310-AWS-Glue-Integration

71cb37e

ashb reviewed Dec 1, 2018

View reviewed changes

Merge branch 'master' into AIRFLOW-2310-AWS-Glue-Integration

5de1774

zingorn reviewed Jan 15, 2019

View reviewed changes

mik-laj reviewed Jan 20, 2019

View reviewed changes

mik-laj reviewed Jan 22, 2019

View reviewed changes

zhongjiajie reviewed May 8, 2019

View reviewed changes

mik-laj added the provider:amazon AWS/Amazon - related issues label Jul 27, 2019

dazza-codes mentioned this pull request Sep 5, 2019

[AIRFLOW-2310] Enable AWS Glue Job Integration #6007

Merged

5 tasks done

kaxil force-pushed the master branch from 558f9f1 to d719e1f Compare October 2, 2019 15:31

stale Bot added the stale Stale PRs per the .github/workflows/stale.yml policy file label Nov 20, 2019

stale Bot closed this Nov 27, 2019

	self.logger.info("Found AWS Glue Job: {job_name}".format(job_name=self.job_name))
	self.logger.info("Found AWS Glue Job: %s", self.job_name)

	super(AwsGlueJobHook, self).__init__(self.aws_conn_id, args, *kwargs)
	super().__init__(self.aws_conn_id, args, *kwargs)

	super(AWSGlueJobOperator, self).__init__(args, *kwargs)
	super().__init__(args, *kwargs)

	super(AwsGlueJobSensor, self).__init__(args, *kwargs)
	super().__init__(args, *kwargs)

Conversation

oelesinsc24 commented Oct 19, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Jira

Description

Tests

Commits

Documentation

Code Quality

Uh oh!

oelesinsc24 commented Oct 19, 2018

Uh oh!

Fokko left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

oelesinsc24 commented Oct 29, 2018

Uh oh!

Fokko commented Nov 5, 2018

Uh oh!

Uh oh!

nwagh commented Nov 15, 2018

Uh oh!

infa-rasrinivasan commented Dec 1, 2018

Uh oh!

oelesinsc24 commented Dec 1, 2018

Uh oh!

ashb left a comment

Choose a reason for hiding this comment

Uh oh!

oelesinsc24 commented Dec 4, 2018

Uh oh!

ashb commented Dec 4, 2018

Uh oh!

shayan90 commented Dec 4, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

oelesinsc24 commented Dec 12, 2018

Uh oh!

ashb commented Dec 12, 2018

Uh oh!

Fokko commented Dec 13, 2018

Uh oh!

oelesinsc24 commented Dec 13, 2018

Uh oh!

Fokko commented Dec 13, 2018

Uh oh!

amelio-vazquez-reina commented Dec 18, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mik-laj Jan 20, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

oelesinsc24 commented Oct 19, 2018 •

edited

Loading

shayan90 commented Dec 4, 2018 •

edited

Loading

mik-laj Jan 20, 2019 •

edited

Loading

dazza-codes commented Aug 29, 2019 •

edited

Loading

dazza-codes commented Aug 29, 2019 •

edited

Loading