Add EMR Serverless Operators and Hooks #25324

syedahsn · 2022-07-26T23:14:42Z

Closes: #20215

Define new operators, as well as necessary hooks and sensors, to run EMR Serverless jobs. System test for EMR Serverless will be coming in a separate PR.

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

Co-Author: https://github.com/dacort

…lures

…as params. Add test to check timeout exception

boring-cyborg · 2022-07-26T23:14:45Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

potiuk · 2022-07-28T18:00:37Z

@ferruzzi @o-nikolas @vincbeck -> WDYT?

vincbeck · 2022-07-28T18:04:05Z

Syed is a new member of our team, we already reviewed his work :) Thanks for reaching out though

potiuk · 2022-07-28T18:06:29Z

Syed is a new member of our team, we already reviewed his work :) Thanks for reaching out though

aha :) Good to know.

airflow/providers/amazon/aws/operators/emr.py

potiuk · 2022-07-29T09:57:08Z

Some errors to fix :)

airflow/providers/amazon/aws/hooks/emr.py

airflow/providers/amazon/aws/operators/emr.py

airflow/providers/amazon/aws/sensors/emr.py

josh-fell · 2022-07-29T14:36:36Z

airflow/providers/amazon/aws/sensors/emr.py

+        **kwargs: Any,
+    ) -> None:
+        self.aws_conn_id = aws_conn_id
+        self.emr_conn_id = emr_conn_id


I could easily be missing it, but where is emr_conn_id being used in this sensor and other sensors in this file?

I agree. It seems emr_conn_id is not used. You can just remove it

airflow/providers/amazon/aws/sensors/emr.py

airflow/providers/amazon/aws/hooks/emr.py

airflow/providers/amazon/aws/sensors/emr.py

vincbeck · 2022-07-29T18:02:35Z

airflow/providers/amazon/aws/sensors/emr.py

+        **kwargs: Any,
+    ) -> None:
+        self.aws_conn_id = aws_conn_id
+        self.emr_conn_id = emr_conn_id


I agree. It seems emr_conn_id is not used. You can just remove it

airflow/providers/amazon/aws/sensors/emr.py

vincbeck · 2022-07-29T18:05:27Z

airflow/providers/amazon/aws/sensors/emr.py

+
+        try:
+            response = self.hook.conn.get_job_run(applicationId=self.application_id, jobRunId=self.job_run_id)
+        except Exception:


Overall question. Do we want or is there any value converting the Exception to AirflowException? I can see there is a lot of operators which do not do that. If there is no value, I would rather not doing the conversion just for the sake of having a code cleaner

I went with Exception just as a catchall for any type of exception that might be thrown by the boto3 API. I'm not sure if there's an advantage to throwing an AirflowException instead of a generic Exception. Only thing I can think of is that it would make it clear that the exception was coming from Airflow, rather than any underlying API calls (i.e. boto3) that might have been made. But that can also be understood from the message associated with the Exception.

FWIW I think AirflowException is used too broadly in the code base. IMO AirflowException, and its subclasses, are really related to Airflow orchestration activity rather than activity done by what Airflow is orchestrating. It might make more sense to drop the try/except and bubble up whatever API call errors that might be thrown directly. Up to you though.

airflow/providers/amazon/aws/sensors/emr.py

docs/apache-airflow-providers-amazon/operators/emr_serverless.rst

dacort · 2022-07-29T18:40:50Z

Closes #20215

potiuk · 2022-08-02T15:53:19Z

some errors need fixing.

- Import cached_property directly functools - Fix indentation in doc string - Define default values in doc string explicitly - Remove unnecessary references to emr_conn_id - Replace print messages with log messages

josh-fell

Some minor suggestions but overall LGTM (after fixing the tests of course).

josh-fell · 2022-08-03T14:51:50Z

docs/apache-airflow-providers-amazon/operators/emr_serverless.rst

+.. _howto/operator:EmrServerlessCreateApplicationOperator:
+
+Create an EMR Serverless Application
+==========================


This underline needs to be at least the length of the text it's underlining. I imagine this might be a Build Docs failure for you.

josh-fell · 2022-08-03T14:56:18Z

airflow/providers/amazon/aws/operators/emr.py

+        self.log.info('Starting job on Application: %s', self.application_id)
+
+        app_state = self.hook.conn.get_application(applicationId=self.application_id)['application']['state']
+        if app_state not in {'CREATED', 'STARTED'}:


Suggested change

if app_state not in {'CREATED', 'STARTED'}:

if app_state not in EmrServerlessApplicationSensor.SUCCESS_STATES:

Small suggestion to use the attrs where you can like you have in other places. Up to you though (assuming it's functionally correct as well).

josh-fell · 2022-08-03T15:01:17Z

airflow/providers/amazon/aws/sensors/emr.py

+
+        try:
+            response = self.hook.conn.get_job_run(applicationId=self.application_id, jobRunId=self.job_run_id)
+        except Exception:


FWIW I think AirflowException is used too broadly in the code base. IMO AirflowException, and its subclasses, are really related to Airflow orchestration activity rather than activity done by what Airflow is orchestrating. It might make more sense to drop the try/except and bubble up whatever API call errors that might be thrown directly. Up to you though.

- Fixed minor formatting in emr_serverless.rst

airflow/providers/amazon/aws/sensors/emr.py

airflow/providers/amazon/aws/hooks/emr.py

docs/apache-airflow-providers-amazon/operators/emr_serverless.rst

eladkal · 2022-08-04T19:17:51Z

airflow/providers/amazon/aws/example_dags/example_emr_serverless.py

@@ -0,0 +1,97 @@
+# Licensed to the Apache Software Foundation (ASF) under one


new example dags should be according to AIP-47
Example: #25205

That's a good point but the plan was to do it in two iterations since this PR is already big enough. Once this PR is merged, @syedahsn will work on converting this sample dag to system test. If you strongly disagree, I guess the two can be done in this PR but it will just make it bigger and harder to review

- Minor formatting in emr_serverless.rst

Seems all addressed.

syedahsn and others added 21 commits July 26, 2022 09:51

Hooks and Operators for EMR Serverless

3ab377c

Remove unused imports

0ee2b1d

Updated work for Dennis

9cdb462

Fix bugs with starting operators

b49bd3d

Add sensor for emr jobs and applications

1fb4758

Linting, Added Missing Docs, and CR Fixes

9ab620d

Fix Fake Waiter

75637be

Co-Author: https://github.com/dacort

EMR dags and docs CR

3fe842f

CR comments rev 2

e8b2f0b

rev 8 changes suggested by Niko

be2954d

Fix typos and documentation for rev 8 comments

95caac2

Move waiter logic to hook, improve sensor functionality to report fai…

d92da3a

…lures

Move wait function to work only when needed

014f89a

Tests for emr serverless

80d3588

Update emr serverless release label to latest version

9e59719

Make changes suggested by Niko

8d41f4c

Mock waiter function call - improve emr serverless hook tests

6cace43

Add support for arbitrary parameters to StartJobOperator

877e79d

Change waiter to make check_interval_seconds and countdown be passed …

5a522e8

…as params. Add test to check timeout exception

Remove duplicate test variables

4d86a59

Remove service.json

471688f

syedahsn requested a review from mik-laj as a code owner July 26, 2022 23:14

boring-cyborg bot added area:providers kind:documentation provider:amazon-aws AWS/Amazon - related issues labels Jul 26, 2022

potiuk reviewed Jul 28, 2022

View reviewed changes

airflow/providers/amazon/aws/operators/emr.py Outdated Show resolved Hide resolved

josh-fell reviewed Jul 29, 2022

View reviewed changes

vincbeck requested changes Jul 29, 2022

View reviewed changes

Make changes requested by reviewers:

8294870

- Import cached_property directly functools - Fix indentation in doc string - Define default values in doc string explicitly - Remove unnecessary references to emr_conn_id - Replace print messages with log messages

josh-fell approved these changes Aug 3, 2022

View reviewed changes

dacort mentioned this pull request Aug 3, 2022

fixed function typos that prevented seeing job errors aws-samples/emr-serverless-samples#17

Merged

Removed redundant try/except around calls to boto3 API

68896f4

- Fixed minor formatting in emr_serverless.rst

ferruzzi reviewed Aug 4, 2022

View reviewed changes

airflow/providers/amazon/aws/sensors/emr.py Outdated Show resolved Hide resolved

josh-fell reviewed Aug 4, 2022

View reviewed changes

airflow/providers/amazon/aws/hooks/emr.py Outdated Show resolved Hide resolved

vincbeck requested changes Aug 4, 2022

View reviewed changes

docs/apache-airflow-providers-amazon/operators/emr_serverless.rst Show resolved Hide resolved

eladkal previously requested changes Aug 4, 2022

View reviewed changes

syedahsn added 2 commits August 4, 2022 17:29

Forgot to remove some print statements

4281433

- Minor formatting in emr_serverless.rst

Add template fields to EmrServerlessStartJobOperator

c60edf1

potiuk requested a review from eladkal August 5, 2022 16:07

potiuk merged commit 8df84e9 into apache:main Aug 5, 2022

This was referenced Aug 10, 2022

Status of testing Providers that were prepared on August 10, 2022 #25634

Closed

Status of testing Providers that were prepared on August 10, 2022 #25640

Closed

dacort mentioned this pull request Aug 13, 2022

Version mismatch in 'airflow/setup.py' aws-samples/emr-serverless-samples#19

Closed

dacort added a commit to aws-samples/emr-serverless-samples that referenced this pull request Aug 15, 2022

Update Airflow code to official PR - apache/airflow#25324

a416d5f

dacort mentioned this pull request Aug 15, 2022

Sync Airflow Operator with official PR aws-samples/emr-serverless-samples#20

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add EMR Serverless Operators and Hooks #25324

Add EMR Serverless Operators and Hooks #25324

syedahsn commented Jul 26, 2022 •

edited by josh-fell

Loading

boring-cyborg bot commented Jul 26, 2022

potiuk commented Jul 28, 2022

vincbeck commented Jul 28, 2022

potiuk commented Jul 28, 2022

potiuk commented Jul 29, 2022

josh-fell Jul 29, 2022

vincbeck Jul 29, 2022

vincbeck Jul 29, 2022

vincbeck Jul 29, 2022

syedahsn Jul 30, 2022

josh-fell Aug 3, 2022

dacort commented Jul 29, 2022

potiuk commented Aug 2, 2022

josh-fell left a comment •

edited

Loading

josh-fell Aug 3, 2022

josh-fell Aug 3, 2022

josh-fell Aug 3, 2022

eladkal Aug 4, 2022

vincbeck Aug 4, 2022

	if app_state not in {'CREATED', 'STARTED'}:
	if app_state not in EmrServerlessApplicationSensor.SUCCESS_STATES:

		@@ -0,0 +1,97 @@
		# Licensed to the Apache Software Foundation (ASF) under one

Add EMR Serverless Operators and Hooks #25324

Add EMR Serverless Operators and Hooks #25324

Conversation

syedahsn commented Jul 26, 2022 • edited by josh-fell Loading

boring-cyborg bot commented Jul 26, 2022

potiuk commented Jul 28, 2022

vincbeck commented Jul 28, 2022

potiuk commented Jul 28, 2022

potiuk commented Jul 29, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dacort commented Jul 29, 2022

potiuk commented Aug 2, 2022

josh-fell left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

syedahsn commented Jul 26, 2022 •

edited by josh-fell

Loading

josh-fell left a comment •

edited

Loading