[AIRFLOW-1349] Refactor BackfillJob _execute #2463

edgarRd · 2017-07-21T02:04:32Z

Dear Airflow maintainers,

Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!

JIRA

My PR addresses the following Airflow JIRA issues and references them in the PR title. For example, "[AIRFLOW-1349] My Airflow PR"
- https://issues.apache.org/jira/browse/AIRFLOW-1349

Description

Here are some details about my PR, including screenshots of any UI changes:

BackfillJob._execute is doing multiple things - it is pretty hard to
follow and maintain.

Changes included are just a re-org of the code, no logic has been
changed.

Refactor includes:

Break BackfillJob._execute into functions
Add a Status object to track BackfillJob internal status while
executing the job.

Tests

My PR adds the following unit tests OR does not need testing for this extremely good reason:
BackfillJobTest.test_backfill_max_limit_check*

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
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"

mention-bot · 2017-07-21T02:04:34Z

@edgarRd, thanks for your PR! By analyzing the history of the files in this pull request, we identified @bolkedebruin, @mistercrunch and @jlowin to be potential reviewers.

codecov-io · 2017-07-21T02:38:03Z

Codecov Report

Merging #2463 into master will increase coverage by 0.09%.
The diff coverage is 91.44%.

@@            Coverage Diff             @@
##           master    #2463      +/-   ##
==========================================
+ Coverage   69.43%   69.53%   +0.09%     
==========================================
  Files         146      146              
  Lines       11351    11391      +40     
==========================================
+ Hits         7882     7921      +39     
- Misses       3469     3470       +1

Impacted Files	Coverage Δ
airflow/models.py	`87.39% <100%> (+0.06%)`	⬆️
airflow/jobs.py	`77.64% <90.78%> (+0.52%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 322ec96...cf4dc3c. Read the comment docs.

bolkedebruin · 2017-07-21T20:00:36Z

airflow/jobs.py

@@ -1790,6 +1790,54 @@ class BackfillJob(BaseJob):
        'polymorphic_identity': 'BackfillJob'
    }

+    class _DagRunTaskStatus(object):


If we are putting this into an class, would it be something to add a integrity check? Ie there is a total number of tasks. The sum of the counters should always be equal to number. And yes we have seen leakage in the past.

Might be outside of this PR though (no logic change).

Agreed on the consistency check, but also agree on the scope creep part. Edgar do you mind adding a TODO and attach it to a JIRA for now?

Done: https://issues.apache.org/jira/browse/AIRFLOW-1444

aoen

LGTM in general mostly nits.

aoen · 2017-07-21T21:25:14Z

airflow/jobs.py

+        it easier to pass it around.
+        """
+
+        # TODO(edgarRd): https://issues.apache.org/jira/browse/AIRFLOW-1444


Nit: let's kill the whitespace/newlines, make it more descriptive (e.g. summary of the JIRA), and no need to include the full JIRA URL, just AIRFLOW-144 is OK (consistent with the other JIRA TODOs in the project)

aoen · 2017-07-21T21:29:40Z

airflow/jobs.py

+                     total_runs=0,
+                     ):
+            """
+            :param started: Dict of tasks already started


Nit: you can remove "Dict of" since it's redundantly given by the type below. Better just explain what the keys/values represent in english.

Same for the other params.

aoen · 2017-07-21T21:32:13Z

airflow/jobs.py

-
-        executor = self.executor
-        executor.start()
+        # explicitly mark as backfill and running as we can fill gaps


What does "as we can fill gaps" mean? I see it's copied from the old logic, let's kill the second part of this sentence if you don't know what it means either.

aoen · 2017-07-21T21:33:02Z

airflow/jobs.py

+    @provide_session
+    def _task_instances_for_dag_run(self, dag_run, session=None):
+        """
+        Returns a map with the tasks to run for the given dag run.


Nit: Good to add what the map is mapping from (type of keys)

aoen · 2017-07-21T21:39:58Z

airflow/jobs.py

-        # Triggering what is ready to get triggered
-        while (len(tasks_to_run) > 0 or len(started) > 0) and not deadlocked:
+        while ((len(tasks_to_run) > 0 or len(ti_status.started) > 0) and
+                not ti_status.deadlocked):


Let's make the ti_status.deadlocked consistent with the other checks in this while loop while we are refactoring it (> 0)

aoen · 2017-07-21T21:51:25Z

airflow/jobs.py

+
+        return err
+
+    def _get_run_dates(self):


I feel like this is generic enough that it better belongs as a public method under the DAG class in models.py, and it can take bf_start_date/end_date as arguments

aoen · 2017-07-21T21:55:25Z

airflow/jobs.py

+        return run_dates
+
+    @provide_session
+    def _execute_for_run_dates(self, run_dates, ti_status, executor, pickle_id,


Would be good to make it more clear what the difference between this method and _execute is (either better method names or comments).

bolkedebruin · 2017-07-23T18:55:49Z

airflow/jobs.py

@@ -1841,40 +1889,39 @@ def __init__(
        self.pool = pool
        super(BackfillJob, self).__init__(*args, **kwargs)

-    def _update_counters(self, started, succeeded, skipped, failed, tasks_to_run):
+    def _update_counters(self, ti_status, tasks_to_run):


Should this not be moved to _DagRunTaskStatus?

Yeah, that makes sense. I've done the change.

saguziel · 2017-07-26T00:20:50Z

airflow/jobs.py

-        Runs a dag for a specified date range.
+        Returns a dag run for the given run date, which will be matched to an existing
+        dag run if available or create a new dag run otherwise.
+        :param run_date: the execution date for the dag run


add :type: annotations too

saguziel · 2017-07-26T00:21:59Z

airflow/jobs.py

+                          execution_date=run_date,
+                          session=session)
+
+        if not run:


It is probably better to check against None

saguziel · 2017-07-26T00:23:00Z

airflow/jobs.py


-            # return if already reached maximum active runs


Maybe a followup change is active runs doesn't count backfills? I'm actually not sure what the current logic is

This is part of what AIRFLOW-1349 will address.

saguziel · 2017-07-26T00:26:23Z

airflow/jobs.py

-            run.state = State.RUNNING
-            run.run_id = run_id
-            run.verify_integrity(session=session)
+        if not dag_run:


prefer to check against None

saguziel · 2017-07-26T00:28:08Z

airflow/jobs.py

+        for ti in dag_run.get_task_instances():
+            # all tasks part of the backfill are scheduled to run
+            if ti.state == State.NONE:
+                ti.set_state(State.SCHEDULED, session=session)


This is pretty slow to run and expires the existing TIs. Better to query as a batch. Keep in mind session.commit expires all attributes

Right, as of this PR I just organized the code to have it easier to make those changes, so this is just a re-org of the code. I could create a ticket to improve the performance here and make a follow up change unless someone else thinks it should be included on this PR @aoen @bolkedebruin

I see. Making a followup change is cool if this is just a refactor

saguziel · 2017-07-26T00:29:34Z

airflow/jobs.py

+        them in a backfill process.
+        :param status: the internal status of the job
+        :type status: BackfillJob._DagRunTaskStatus
+        :param executor: the executor to run the task instances


edgarRd · 2017-07-27T01:09:17Z

@aoen @bolkedebruin @saguziel is there any other comment / issue in order to move this PR forward?

aoen · 2017-07-28T20:38:01Z

airflow/jobs.py

+            :type not_ready: set
+            :param deadlocked: Deadlocked tasks
+            :type deadlocked: set
+            :param active_runs: Active tasks at a certain point in time (mutate)


What does mutate mean? Let's clarify or kill

aoen · 2017-07-28T20:39:49Z

airflow/jobs.py

+    def _execute_for_run_dates(self, run_dates, ti_status, executor, pickle_id,
+                               start_date, session=None):
+        """
+        This method does the bulk of the work, receiving all initialized components ready


Nit: bulk of the work doesn't explain what work.

BackfillJob._execute is doing multiple things - it is pretty hard to follow and maintain. Changes included are just a re-org of the code, no logic has been changed. Refactor includes: - Break BackfillJob._execute into functions - Add a Status object to track BackfillJob internal status while executing the job.

BackfillJob._execute is doing multiple things - it is pretty hard to follow and maintain. Changes included are just a re-org of the code, no logic has been changed. Refactor includes: - Break BackfillJob._execute into functions - Add a Status object to track BackfillJob internal status while executing the job. Closes apache#2463 from edgarRd/erod-backfill-refactor

bolkedebruin reviewed Jul 21, 2017

View reviewed changes

edgarRd force-pushed the erod-backfill-refactor branch from b70f175 to 4a1d500 Compare July 21, 2017 21:21

aoen reviewed Jul 21, 2017

View reviewed changes

edgarRd force-pushed the erod-backfill-refactor branch from 4a1d500 to b6e2e75 Compare July 22, 2017 00:13

bolkedebruin reviewed Jul 23, 2017

View reviewed changes

edgarRd force-pushed the erod-backfill-refactor branch from b6e2e75 to 124568c Compare July 24, 2017 21:14

saguziel reviewed Jul 26, 2017

View reviewed changes

edgarRd force-pushed the erod-backfill-refactor branch from 124568c to cb3b3a4 Compare July 26, 2017 01:43

aoen reviewed Jul 28, 2017

View reviewed changes

edgarRd force-pushed the erod-backfill-refactor branch from cb3b3a4 to cf4dc3c Compare July 28, 2017 20:52

asfgit closed this in 547f818 Jul 28, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[AIRFLOW-1349] Refactor BackfillJob _execute #2463

[AIRFLOW-1349] Refactor BackfillJob _execute #2463

edgarRd commented Jul 21, 2017

mention-bot commented Jul 21, 2017

codecov-io commented Jul 21, 2017 •

edited

bolkedebruin Jul 21, 2017

aoen Jul 21, 2017

edgarRd Jul 21, 2017

aoen left a comment

aoen Jul 21, 2017

aoen Jul 21, 2017

aoen Jul 21, 2017

aoen Jul 21, 2017

aoen Jul 21, 2017

aoen Jul 21, 2017

aoen Jul 21, 2017

bolkedebruin Jul 23, 2017

edgarRd Jul 24, 2017

saguziel Jul 26, 2017

edgarRd Jul 26, 2017

saguziel Jul 26, 2017

edgarRd Jul 26, 2017

saguziel Jul 26, 2017

edgarRd Jul 26, 2017

saguziel Jul 26, 2017

edgarRd Jul 26, 2017

saguziel Jul 26, 2017

edgarRd Jul 26, 2017

saguziel Jul 26, 2017

saguziel Jul 26, 2017

edgarRd Jul 26, 2017

edgarRd commented Jul 27, 2017

aoen Jul 28, 2017

aoen Jul 28, 2017

[AIRFLOW-1349] Refactor BackfillJob _execute #2463

[AIRFLOW-1349] Refactor BackfillJob _execute #2463

Conversation

edgarRd commented Jul 21, 2017

JIRA

Description

Tests

Commits

mention-bot commented Jul 21, 2017

codecov-io commented Jul 21, 2017 • edited

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aoen left a 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

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

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

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

edgarRd commented Jul 27, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-io commented Jul 21, 2017 •

edited