You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2020/09/10 00:10:26 UTC
[GitHub] [airflow] tszerszen opened a new pull request #10847: Add on_kill method to DataprocSubmitJobOperator
tszerszen opened a new pull request #10847:
URL: https://github.com/apache/airflow/pull/10847
This is my proposition of solving issue [#10381](https://github.com/apache/airflow/issues/10381)
Class `DataprocSubmitJobOperator` in package `airflow.providers.google.cloud.operators.dataproc` has now `on_kill` method. Since canceling job requires access to `DataprocHook` and `job_id` I added `hook` and `job_id` as properties in `DataprocSubmitJobOperator` class.
Tests:
Class `TestDataprocSubmitJobOperator` in package `tests.providers.google.cloud.operators.test_dataproc` has now `test_on_kill` method.
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486150151
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1834,15 +1838,24 @@ def execute(self, context: Dict):
timeout=self.timeout,
metadata=self.metadata,
)
- job_id = job_object.reference.job_id
- self.log.info('Job %s submitted successfully.', job_id)
+ self.job_id = job_object.reference.job_id
+ self.log.info('Job %s submitted successfully.', self.job_id)
if not self.asynchronous:
- self.log.info('Waiting for job %s to complete', job_id)
- hook.wait_for_job(job_id=job_id, location=self.location, project_id=self.project_id)
- self.log.info('Job %s completed successfully.', job_id)
+ self.log.info('Waiting for job %s to complete', self.job_id)
+ self.hook.wait_for_job(job_id=self.job_id, location=self.location, project_id=self.project_id)
+ self.log.info('Job %s completed successfully.', self.job_id)
+
+ return self.job_id
- return job_id
+ def on_kill(self, cancel_on_kill: bool = True, job_id: str = ''):
Review comment:
```suggestion
def on_kill(self):
```
This is an abstract method and those parameters will never be passed. We have to use operators attributes
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek merged pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
turbaszek merged pull request #10847:
URL: https://github.com/apache/airflow/pull/10847
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tszerszen commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486111600
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1808,6 +1808,8 @@ def __init__(
gcp_conn_id: str = "google_cloud_default",
impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
asynchronous: bool = False,
+ hook: Optional[DataprocHook] = None,
Review comment:
@michalslowikowski00 > Why do you want to create hook object within the `__init__` method?
I wanted `DataprocSubmitJobOperator` to be more customizable, and I need `hook` and `job_id` as properties to be able to cancel job in `on_kill` method.
If that's better I can disallow to initialise these properties in `__init__`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] michalslowikowski00 commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
michalslowikowski00 commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486106269
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1808,6 +1808,8 @@ def __init__(
gcp_conn_id: str = "google_cloud_default",
impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
asynchronous: bool = False,
+ hook: Optional[DataprocHook] = None,
Review comment:
Why do you want to create hook object within the `__init__` method?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486150692
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1834,15 +1838,24 @@ def execute(self, context: Dict):
timeout=self.timeout,
metadata=self.metadata,
)
- job_id = job_object.reference.job_id
- self.log.info('Job %s submitted successfully.', job_id)
+ self.job_id = job_object.reference.job_id
+ self.log.info('Job %s submitted successfully.', self.job_id)
if not self.asynchronous:
- self.log.info('Waiting for job %s to complete', job_id)
- hook.wait_for_job(job_id=job_id, location=self.location, project_id=self.project_id)
- self.log.info('Job %s completed successfully.', job_id)
+ self.log.info('Waiting for job %s to complete', self.job_id)
+ self.hook.wait_for_job(job_id=self.job_id, location=self.location, project_id=self.project_id)
+ self.log.info('Job %s completed successfully.', self.job_id)
+
+ return self.job_id
- return job_id
+ def on_kill(self, cancel_on_kill: bool = True, job_id: str = ''):
+ if not cancel_on_kill:
Review comment:
```suggestion
if not self.cancel_on_kill:
```
This param should be added to operator constructor and assigned ther as operator attribute `self.cancel_on_kill = cancel_on_kill`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486151408
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1834,15 +1838,24 @@ def execute(self, context: Dict):
timeout=self.timeout,
metadata=self.metadata,
)
- job_id = job_object.reference.job_id
- self.log.info('Job %s submitted successfully.', job_id)
+ self.job_id = job_object.reference.job_id
+ self.log.info('Job %s submitted successfully.', self.job_id)
if not self.asynchronous:
- self.log.info('Waiting for job %s to complete', job_id)
- hook.wait_for_job(job_id=job_id, location=self.location, project_id=self.project_id)
- self.log.info('Job %s completed successfully.', job_id)
+ self.log.info('Waiting for job %s to complete', self.job_id)
+ self.hook.wait_for_job(job_id=self.job_id, location=self.location, project_id=self.project_id)
+ self.log.info('Job %s completed successfully.', self.job_id)
+
+ return self.job_id
- return job_id
+ def on_kill(self, cancel_on_kill: bool = True, job_id: str = ''):
+ if not cancel_on_kill:
+ return
+ cancel_job_id = self.job_id
+ if job_id:
+ cancel_job_id = job_id
+ if self.hook and cancel_job_id:
+ self.hook.cancel_job(job_id=self.job_id, project_id=self.project_id, location=self.location)
Review comment:
```suggestion
if self. job_id and self.cancel_on_kill:
self.hook.cancel_job(job_id=self.job_id, project_id=self.project_id, location=self.location)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486148270
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1808,6 +1808,8 @@ def __init__(
gcp_conn_id: str = "google_cloud_default",
impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
asynchronous: bool = False,
+ hook: Optional[DataprocHook] = None,
Review comment:
The general idea is that operators should initialize the hook in execute methods. What users should pass in operators constructor is parameters for hook initialization (in this case `gcp_conn_id` and `impersonation_chain`).
This operator can use only `DataprocHook` using any other hook may result in strange errors. Recommended way to to create "custom behavior" in Airflow is creating new operators.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486241720
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1808,6 +1808,7 @@ def __init__(
gcp_conn_id: str = "google_cloud_default",
impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
asynchronous: bool = False,
+ cancel_on_kill: bool = True,
Review comment:
Can you add please this parameter to docstring so user can understand what it does?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486240748
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1834,15 +1838,19 @@ def execute(self, context: Dict):
timeout=self.timeout,
metadata=self.metadata,
)
- job_id = job_object.reference.job_id
- self.log.info('Job %s submitted successfully.', job_id)
+ self.job_id = job_object.reference.job_id
+ self.log.info('Job %s submitted successfully.', self.job_id)
if not self.asynchronous:
- self.log.info('Waiting for job %s to complete', job_id)
- hook.wait_for_job(job_id=job_id, location=self.location, project_id=self.project_id)
- self.log.info('Job %s completed successfully.', job_id)
+ self.log.info('Waiting for job %s to complete', self.job_id)
+ self.hook.wait_for_job(job_id=self.job_id, location=self.location, project_id=self.project_id)
+ self.log.info('Job %s completed successfully.', self.job_id)
+
+ return self.job_id
Review comment:
Looks good! What would you say to limit number of changes by adding `self.job_id = job_id` before return?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#issuecomment-690523633
Awesome work, congrats on your first merged pull request!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tszerszen commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486108435
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1808,6 +1808,8 @@ def __init__(
gcp_conn_id: str = "google_cloud_default",
impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
asynchronous: bool = False,
+ hook: Optional[DataprocHook] = None,
+ job_id: str = '',
Review comment:
@michalslowikowski00 > I think we always use such notation.
I think such change will raise `SyntaxError: non-default argument follows default argument`, since `job_id` is optional.
What about the following notation?
```
job_id: Optional[str] = None,
```
But I got mypy error with the above, becouse `job_id` might be `None` when method `cancel_job` on `hook` is called.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] michalslowikowski00 commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
michalslowikowski00 commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486104628
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1808,6 +1808,8 @@ def __init__(
gcp_conn_id: str = "google_cloud_default",
impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
asynchronous: bool = False,
+ hook: Optional[DataprocHook] = None,
+ job_id: str = '',
Review comment:
I think we always use such notation.
```suggestion
job_id: str,
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#issuecomment-689890146
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/master/CONTRIBUTING.rst)
Here are some useful points:
- Pay attention to the quality of your code (flake8, pylint and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/master/STATIC_CODE_CHECKS.rst#prerequisites-for-pre-commit-hooks) 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](https://github.com/apache/airflow/blob/master/docs/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/master/BREEZE.rst) 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](https://www.apache.org/foundation/policies/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]( https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#coding-style-and-best-practices).
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://apache-airflow-slack.herokuapp.com/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486149606
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1821,11 +1823,13 @@ def __init__(
self.gcp_conn_id = gcp_conn_id
self.impersonation_chain = impersonation_chain
self.asynchronous = asynchronous
+ self.hook = hook
+ self.job_id = job_id
Review comment:
```suggestion
self.hook: Optional[DataprocHook] = None
self.job_id: Optional[str] = None
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tszerszen commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486241624
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1821,11 +1823,13 @@ def __init__(
self.gcp_conn_id = gcp_conn_id
self.impersonation_chain = impersonation_chain
self.asynchronous = asynchronous
+ self.hook = hook
+ self.job_id = job_id
Review comment:
`self.job_id: Optional[str] = None` couses mypy error:
```
airflow/providers/google/cloud/operators/dataproc.py:1846: error: Argument
"job_id" to "wait_for_job" of "DataprocHook" has incompatible type
"Optional[str]"; expected "str"
self.hook.wait_for_job(job_id=self.job_id, location=self.l...
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10847: Add on_kill method to DataprocSubmitJobOperator
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10847:
URL: https://github.com/apache/airflow/pull/10847#discussion_r486149116
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -1808,6 +1808,8 @@ def __init__(
gcp_conn_id: str = "google_cloud_default",
impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
asynchronous: bool = False,
+ hook: Optional[DataprocHook] = None,
+ job_id: str = '',
Review comment:
```suggestion
```
No need for this one. The `job_id` should be assigned in `execute` method that is executed before `on_kill`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org