feluelle commented on a change in pull request #9022:
URL: https://github.com/apache/airflow/pull/9022#discussion_r454163716
##########
File path: airflow/providers/amazon/aws/sensors/glue.py
##########
@@ -47,14 +47,13 @@ def __init__(self,
self.aws_conn_id = aws_conn_id
self.success_states = ['SUCCEEDED']
self.errored_states = ['FAILED', 'STOPPED', 'TIMEOUT']
+ self.hook = AwsGlueJobHook(aws_conn_id=self.aws_conn_id)
Review comment:
A hook should not be initialzed during operator/sensor init, because it
will be called everytime the webserver parses the dag file the operator is
initiated (`task = AwsGlueJobSensor(...)`). The hook initialisation sometimes
contains connections to external services which would slow down the Airflow
instance.
You only need the hook object in the `poke`/`execute` function when you
actually run the task.
##########
File path: airflow/providers/amazon/aws/hooks/glue.py
##########
@@ -110,24 +110,36 @@ def initialize_job(self, script_arguments=None):
self.log.error("Failed to run aws glue job, error: %s",
general_error)
raise
- def job_completion(self, job_name=None, run_id=None):
+ def get_job_state(self, job_name=None, run_id=None):
"""
:param job_name: unique job name per AWS account
:type job_name: str
:param run_id: The job-run ID of the predecessor job run
:type run_id: str
:return: Status of the Job if succeeded or stopped
"""
+ glue_client = self.get_conn()
+ job_status = glue_client.get_job_run(
+ JobName=job_name,
+ RunId=run_id,
+ PredecessorsIncluded=True
+ )
+ job_run_state = job_status['JobRun']['JobRunState']
+ return job_run_state
+
+ def job_completion(self, job_name=None, run_id=None):
+ """
+ :param job_name: unique job name per AWS account
Review comment:
Same here, please :)
##########
File path: airflow/providers/amazon/aws/hooks/glue.py
##########
@@ -110,24 +110,36 @@ def initialize_job(self, script_arguments=None):
self.log.error("Failed to run aws glue job, error: %s",
general_error)
raise
- def job_completion(self, job_name=None, run_id=None):
+ def get_job_state(self, job_name=None, run_id=None):
"""
:param job_name: unique job name per AWS account
Review comment:
Please also add a short title / description what this function does. :)
##########
File path: airflow/providers/amazon/aws/hooks/glue.py
##########
@@ -110,24 +110,36 @@ def initialize_job(self, script_arguments=None):
self.log.error("Failed to run aws glue job, error: %s",
general_error)
raise
- def job_completion(self, job_name=None, run_id=None):
+ def get_job_state(self, job_name=None, run_id=None):
"""
:param job_name: unique job name per AWS account
:type job_name: str
:param run_id: The job-run ID of the predecessor job run
:type run_id: str
:return: Status of the Job if succeeded or stopped
Review comment:
That is not correct here. `get_job_state` can return any state?!
----------------------------------------------------------------
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:
[email protected]
