[AIRFLOW-160] Parse DAG files through child processes #1636

plypaul · 2016-06-30T00:17:13Z

This is a follow up to #1559 - we've tested #1559 in production for several weeks and now we're looking to get it into master.

Now that the SCHEDULED state has been introduced, tasks start out in the SCHEDULED state and once the task is sent to the executor, it is set to the QUEUED state. Tagging @bolkedebruin for thoughts.

aoen · 2016-06-30T00:37:47Z

airflow/jobs.py

Will remove.

bolkedebruin · 2016-06-30T08:28:11Z

Cool! Im trying to have it tested over the weekend or early next week. Great work. Still wondering about the different states and if it stays coherent. Would you mind updating the scheduler wiki for it?

bolkedebruin · 2016-07-02T18:53:25Z

airflow/jobs.py

Is this really required to do this here? I personally like a more centralized way of configuring logging, maybe not even output it to a file at all but send it to a centralized logging facility.

For debugging, we often need to look at how a DAG file scheduled and the associated logging messages. We found this approach to be convenient because otherwise, the main log file becomes a jumble with many processes running in parallel.

bolkedebruin · 2016-07-02T19:18:36Z

@plypaul

Now that the SCHEDULED state has been introduced, tasks start out in the SCHEDULED state and once the task is sent to the executor, it is set to the QUEUED state.

is it your intention to have QUEUED being set by the executor and not by the scheduler? Now the scheduler seems to set two different exit states for TaskInstances (actually 3 with UP_FOR_RETRY).

plypaul · 2016-07-06T22:34:54Z

I was in favor of changing the state in the scheduler instead of the executor so that state changes are more centralized.

plypaul · 2016-07-06T23:58:46Z

Let me update the wiki once we finalize these changes...

ghost · 2016-07-11T00:56:14Z

airflow/jobs.py

I think you meant len(dag_runs) == 0 here :)

Yep, good catch.

bolkedebruin · 2016-07-12T19:38:14Z

airflow/jobs.py

.format to stay in line with the rest of airfow (and below)?

Instead of parsing the DAG definition files in the same process as the scheduler, this change parses the files in a child process. This helps to isolate the scheduler from bad user code.

plypaul · 2016-07-27T23:06:25Z

@bolkedebruin - checking up since it's been a while

bolkedebruin · 2016-07-28T08:52:15Z

Hey @plypaul sorry for the delay something to do with summer and not being behind a computer. On the delay from queued to running this was just a hunch (sorry if I put you on the wrong path). It just seemed to take longer than I expected purely subjectively.

Would you mind explaining the reasoning for choosing queued vs scheduled?

plypaul · 2016-07-28T21:56:47Z

When the DAG is parsed and the scheduler figures out that a task instance needs to run, it's placed into the SCHEDULED state. Only tasks in the SCHEDULED state get sent to the executor. Once the task gets sent to the executor, it's placed into the QUEUED state. Since only tasks in the SCHEDULED state get sent to the executor, it eliminates the problem of the same task getting sent to the executor multiple times.

bolkedebruin · 2016-07-28T22:03:13Z

Is the queued state set by the executor or by the scheduler? (Reading from phone)

plypaul · 2016-07-28T22:50:20Z

It's set by the scheduler.

bolkedebruin · 2016-07-29T05:52:01Z

Ok LGTM, I do think some bits need to be ironed out but we can figure that out when it is in.

bolkedebruin · 2016-08-06T16:41:18Z

@plypaul just a note:

            # There should only be one DAG run. Add some logging info if this
            # is not the case for later debugging.

might not be true anymore if the backfill dagrun PR gets merged. Due to "lineage" and "auditing" protection: then this can happen in its first iteration as we are not using the real primary of the dag runs

mtdewulf · 2016-08-11T21:18:31Z

airflow/models.py

+        # Keeping existing logic, but not entirely sure why this is here.
+        if not pickle_id and dag:
+            if dag.full_filepath != dag.filepath:
+                path = "DAGS_FOLDER/{}".format(dag.filepath)


I'm late to the party, but I think there should be a path = None initialization above?

You are right, I merged a PR to fix this issue yesterday.

bolkedebruin · 2016-08-20T15:05:23Z

@plypaul can you please update the documentation on this? I would like to add some architectural fixes, but I have some trouble going around in the code.

r39132 · 2016-08-20T16:55:30Z

@plypaul
One question. If most folks have one dag per file, is scheduling potentially slower? I will say that running the sequential executor with the new scheduler does not appear to work after say 5-10 minutes. It seems to get into a state where nothing gets scheduled.

Pull the latest code and unpause (or enable) 7+ examples. You will see it work for a short while and then nothing gets scheduled anymore. It works fine with LocalExecutor (and I suspect Celery, since you folks are running that).

plypaul · 2016-08-22T08:48:14Z

@bolkedebruin Yep, just recently got wiki access. Let me get to that on Monday.

@r39132 Since there are multiple processes, and each process schedules one file, scheduling should be distributed and fast. It's strange that the SequentialExecutor is slow. During testing, I used the SequentialExecutor and did not run into any issues. I'll try it again this week.

plypaul · 2016-08-24T01:08:10Z

@bolkedebruin Posted an updated description at https://cwiki.apache.org/confluence/display/AIRFLOW/Scheduler+Basics

Unfortunately, I have a bit of a backlog as I am taking care of tasks before heading offline from 8/26 - 9/3. I'm not entirely sure I'll be able to get to the SequentialExecutor testing in the next couple of days, but I'll update here if I can.

bolkedebruin · 2016-08-24T08:55:07Z

@plypaul Thanks. Unfortunately there seem to be more issues. There is quantitatively a large delay in tasks getting "run". I can reproduce the behavior quite easily and we have several reports (mailinglist, gitter) where people think tasks are not getting scheduled at all. Hopefully we can spend a little bit of time on this before your holiday so either we fix it before or I can help fixing it (by knowing where to look) .

(Tested on LovalExecutor btw)

Also: @r39132 @jlowin @mistercrunch

plypaul · 2016-08-25T01:18:46Z

@bolkedebruin So I pulled the latest code from master and tried it out with a test DAG using the LocalExecutor, and it seems to be running without unexpected delays for me. I'm wondering if the issue you're seeing is related to the minimum DAG file processing interval that was introduced in this PR. The default value is 3 minutes, which in retrospect, seems long.

I put out a PR (#1761) to change the default so that it runs as fast as possible.

bolkedebruin · 2016-08-25T06:11:28Z

@plypaul it might be, however I am observing it in between scheduler loops as well and I hope that that is not affected by this setting. I'll test it asap

Instead of parsing the DAG definition files in the same process as the scheduler, this change parses the files in a child process. This helps to isolate the scheduler from bad user code. Closes apache#1636 from plypaul/plypaul_schedule_by_file_rebase_master

plypaul force-pushed the plypaul_schedule_by_file_rebase_master branch 2 times, most recently from d2ba98e to c9b5fa2 Compare June 30, 2016 00:30

aoen reviewed Jun 30, 2016
View reviewed changes

plypaul force-pushed the plypaul_schedule_by_file_rebase_master branch 3 times, most recently from 336e71d to 7093520 Compare July 2, 2016 01:20

bolkedebruin reviewed Jul 2, 2016
View reviewed changes

plypaul force-pushed the plypaul_schedule_by_file_rebase_master branch 3 times, most recently from eeedd0b to e03d12d Compare July 7, 2016 22:51

ghost reviewed Jul 11, 2016
View reviewed changes

plypaul force-pushed the plypaul_schedule_by_file_rebase_master branch 4 times, most recently from 0696f16 to e347b68 Compare July 12, 2016 19:34

bolkedebruin reviewed Jul 12, 2016
View reviewed changes

airflow/jobs.py Outdated

Copy link

Contributor

bolkedebruin Jul 12, 2016 •

edited

Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

.format to stay in line with the rest of airfow (and below)?

plypaul force-pushed the plypaul_schedule_by_file_rebase_master branch 2 times, most recently from 726edf5 to a9ef6ff Compare July 13, 2016 03:13

plypaul force-pushed the plypaul_schedule_by_file_rebase_master branch 3 times, most recently from 036edb4 to dd97758 Compare July 26, 2016 20:19

[AIRFLOW-160] Parse DAG files through child processes

a83ca70

Instead of parsing the DAG definition files in the same process as the scheduler, this change parses the files in a child process. This helps to isolate the scheduler from bad user code.

plypaul force-pushed the plypaul_schedule_by_file_rebase_master branch 5 times, most recently from a31a20e to a4d9afc Compare July 27, 2016 01:18

Additional commit (will squash later)

92ac293

plypaul force-pushed the plypaul_schedule_by_file_rebase_master branch from a4d9afc to 92ac293 Compare July 27, 2016 01:38

asfgit closed this in fdb7e94 Jul 31, 2016

mtdewulf reviewed Aug 11, 2016
View reviewed changes

[AIRFLOW-160] Parse DAG files through child processes #1636

[AIRFLOW-160] Parse DAG files through child processes #1636

Uh oh!

Conversation

plypaul commented Jun 30, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bolkedebruin commented Jun 30, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bolkedebruin commented Jul 2, 2016

Uh oh!

plypaul commented Jul 6, 2016

Uh oh!

plypaul commented Jul 6, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bolkedebruin Jul 12, 2016 • 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!

plypaul commented Jul 27, 2016

Uh oh!

bolkedebruin commented Jul 28, 2016

Uh oh!

plypaul commented Jul 28, 2016

Uh oh!

bolkedebruin commented Jul 28, 2016

Uh oh!

plypaul commented Jul 28, 2016

Uh oh!

bolkedebruin commented Jul 29, 2016

Uh oh!

bolkedebruin commented Aug 6, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bolkedebruin commented Aug 20, 2016

Uh oh!

r39132 commented Aug 20, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

plypaul commented Aug 22, 2016

Uh oh!

plypaul commented Aug 24, 2016

Uh oh!

bolkedebruin commented Aug 24, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

plypaul commented Aug 25, 2016

Uh oh!

bolkedebruin commented Aug 25, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

bolkedebruin Jul 12, 2016 •

edited

Loading

r39132 commented Aug 20, 2016 •

edited

Loading

bolkedebruin commented Aug 24, 2016 •

edited

Loading