[AIRFLOW-6181] Add InProcessExecutor #6740

turbaszek · 2019-12-05T17:54:53Z

Make sure you have checked all steps below.

Jira

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

Description

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

Together with guys from Databand we created a new executor that is meant to be used mainly for debugging and DAG development purposes. This executor executes single task instance at time and is able to work with SQLite and sensors.

Using this executor you can debug your DAGs from IDE 🚀

Tests

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

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 (not including Jira issue reference)
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"

Documentation

In case of new functionality, my PR adds documentation that describes how to use it.
- All the public functions and the classes in the PR contain docstrings that explain what it does
- If you implement backwards incompatible changes, please leave a note in the Updating.md so we can assign it to a appropriate release

docs/executor/inprocess.rst

codecov-io · 2019-12-05T20:02:05Z

Codecov Report

Merging #6740 into master will decrease coverage by 0.06%.
The diff coverage is 90.9%.

@@            Coverage Diff            @@
##           master   #6740      +/-   ##
=========================================
- Coverage   84.37%   84.3%   -0.07%     
=========================================
  Files         672     673       +1     
  Lines       38214   38350     +136     
=========================================
+ Hits        32242   32330      +88     
- Misses       5972    6020      +48

Impacted Files	Coverage Δ
airflow/configuration.py	`93.11% <ø> (ø)`	⬆️
airflow/executors/base_executor.py	`96.05% <100%> (ø)`	⬆️
airflow/executors/executor_loader.py	`78.94% <100%> (+15.9%)`	⬆️
airflow/models/taskinstance.py	`93.75% <50%> (-0.27%)`	⬇️
airflow/executors/debug_executor.py	`92.06% <92.06%> (ø)`
airflow/kubernetes/volume_mount.py	`44.44% <0%> (-55.56%)`	⬇️
airflow/kubernetes/volume.py	`52.94% <0%> (-47.06%)`	⬇️
airflow/kubernetes/pod_launcher.py	`45.25% <0%> (-46.72%)`	⬇️
airflow/kubernetes/refresh_config.py	`50.98% <0%> (-23.53%)`	⬇️
...rflow/contrib/operators/kubernetes_pod_operator.py	`78.2% <0%> (-20.52%)`	⬇️
... and 10 more

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 479ee63...fbae6ac. Read the comment docs.

KevinYang21 · 2019-12-05T21:27:41Z

Do we expect people to use the production DB together with this executor? Seems to be quite dangerous. If we expect people to setup local meta DB, do we want to call it out and maybe somehow force to use that local meta DB?

mik-laj · 2019-12-05T21:42:09Z

How do you want to force the local database? In the world of containers it is very difficult to distinguish between a local and a remote database.

turbaszek · 2019-12-05T21:53:41Z

I am against forcing any db. The executor is meant for local, development purposes meaning that there should be no production db to mess with. The way I would expect DAG creators to use it is to go with local environment + sqlite (or any other) or use breeze / other image as their environment.

TESTING.rst

airflow/models/taskinstance.py

docs/executor/inprocess.rst

airflow/executors/executor_loader.py

potiuk · 2019-12-07T21:49:25Z

Do we expect people to use the production DB together with this executor? Seems to be quite dangerous. If we expect people to setup local meta DB, do we want to call it out and maybe somehow force to use that local meta DB?

I think it's exactly the same case as with SequentialExecutor. I consider the executor choice as actually part of deployment (I.e. some people use LocalExecutor, some Celery some Kubernetes)
So i do not see any danger with this executor

potiuk · 2019-12-07T22:10:49Z

airflow/executors/executor_loader.py

+        if executor_name in executors:
+            executor_module = importlib.import_module(executors[executor_name])
+            executor = getattr(executor_module, executor_name)
+            return executor()


I agree. this is muuuch cleaner.

potiuk · 2019-12-07T22:12:25Z

Hwey @kaxil @ashb - you might want to take a look. This is the new InProcessExecutor contributed by Databand.ai and perfected by @nuclearpinguin we told you about. It's really great for testing/debugging of DAGs.

potiuk · 2019-12-09T11:35:44Z

I'd go for it. There is low risk it will break anything and I think it is super useful for anyone testing DAGs. Maybe also we should announce it in devlist/slack the there is this new way of running the in-process executor. I would love to cherry-pick all those related changes (pylint & others) to 1.10.7 as soon as possible.

potiuk · 2019-12-09T11:35:56Z

@mik-laj ?

TESTING.rst

airflow/executors/inprocess_executor.py

ashb · 2019-12-09T16:55:51Z

My main thought is where is this useful, that airflow test isn't? I remember reading somewhere about using airflow test as the entrypoint for IDE.

ashb · 2019-12-09T16:58:11Z

TESTING.rst

+
+    if __name__ == '__main__':
+      dag.clear(reset_dag_runs=True)
+      dag.run()


Isn't there already a dag.cli() or something?

Yes, there is. I would like us to tie in to that command somehow please.

dag.cli() does not work as far as I know (planned to fix). I was thinking about adding debug method, so only this has to be done in DAG, no additional configuration of running a file, WDYT?

airflow/models/taskinstance.py

potiuk · 2019-12-09T17:13:44Z

My main thought is where is this useful, that airflow test isn't? I remember reading somewhere about using airflow test as the entrypoint for IDE.

In all other executors the main process is running subprocess.popen. That prevents python debuggers from debugging then new process forked as the result. So you cannot set breakpoint and hit "Debug" to get it working. The only way we could find so far was remote debugging, but it requires paid version of IntelliJ and it is rather complex to run.

With this setting you go back to a "sane" way of debugging DAGs -you just add those two lines, set the breakpoint and use "Debug". I am not aware of any simple way of doing with with starting subprocesses. Unless someone knows it (but I have not heard it so far).

Moreover with this we will also be able to debug the code inside Breeze using Docker integration (in the same super-intuitive and easy way). You just add two lines to your DAG, set your environment to point to your Docker image/container and you can use all the debugging features of your IDE out of the box to not only initialize but also execute your DAG. This is super-powerful. We are going to use it at our workshops we have this Friday and this is so much easier for the users to debug the DAGs this way. Even if you use other IDEs which have good debugging integration, it's going to be super easy because you debug it in exactly the same way as you debug other python programs (which means it will just work).

I know @feluelle tried it before but could not make it work with the other executors.

ashb · 2019-12-09T19:33:05Z

airflow test doesn't run an executor -- it runs a task directly.

(sorry for short replies -- will expand on this tomorrow)

airflow/executors/executor_loader.py

ashb · 2019-12-09T19:42:31Z

airflow/executors/inprocess_executor.py

+    sqlite since sqlite do not support multiple connections.
+
+    It executes one task instance at time. Additionally to support working
+    with sensors, sensor's ``mode`` will be automatically set to "reschedule".


Why? This requirement/hard-coding doesn't make immediate sense to me.

Suggested change

with sensors, sensor's ``mode`` will be automatically set to "reschedule".

with sensors, all sensors ``mode`` will be automatically set to "reschedule".

The problem is that if sensor depends on another task doing it's work, it will block the executor and will not let the operator to do its job. We had a few examples of those when we worked on GCP operators. Basically - when you want to spin-off both sensor and operator at the same time you need this. Example here:

https://github.com/apache/airflow/blob/master/airflow/gcp/example_dags/example_bigtable.py

Here we have two sensors waiting and they might be fired in any order - they are waiting for two creates (they trigger the sensor's replicate wait to complete). We have more than one sensor and they can fire in any order as they are not depending on each other.

Using "reschedule" mode makes it robust to more strict dependencies set in the DAG.

airflow/executors/inprocess_executor.py

ashb

I think this should be called something like DebugExecutor -- InProcess makes sense to us as developers of Airflow, but might notbe immedately obvious to users. Plus Debug make is more obvious that you shouldn't be using it in production.

potiuk · 2019-12-09T20:40:54Z

airflow test doesn't run an executor -- it runs a task directly.

(sorry for short replies -- will expand on this tomorrow)

'airflow test' only runs one task but then we want to run the whole dag - sometimes when you have a complex DAG you want to run all steps before - and running them manually is not a good idea. you need to run them in the right sequence manually. Being able to run the whole DAG is much more convenient.

airflow/models/taskinstance.py

docs/executor/debug.rst

dimberman · 2019-12-11T19:51:35Z

airflow/config_templates/default_airflow.cfg

+[debug]
+# Used only with DebugExecutor. If set to True DAG will fail with first
+# failed task. Helpful for debugging purposes.
+fail_fast = False


dimberman

Minor nit but overall this LGTM!

tests/executors/test_inprocess_executor.py

potiuk · 2019-12-12T09:34:01Z

I think we should merge it after travis passes :)

Adds new executor that is meant to be used mainly for debugging and DAG development purposes. This executor executes single task instance at time and is able to work with SQL Lite and sensors.

dimberman · 2019-12-12T17:21:07Z

@nuclearpinguin I'll merge as soon as tests pass :)

Adds new executor that is meant to be used mainly for debugging and DAG development purposes. This executor executes single task instance at time and is able to work with SQLLite and sensors. (cherry picked from commit fe2334f)

Adds new executor that is meant to be used mainly for debugging and DAG development purposes. This executor executes single task instance at time and is able to work with SQLLite and sensors.

mik-laj reviewed Dec 5, 2019

View reviewed changes

docs/executor/inprocess.rst Outdated Show resolved Hide resolved

potiuk requested changes Dec 6, 2019

View reviewed changes

feluelle reviewed Dec 6, 2019

View reviewed changes

airflow/executors/executor_loader.py Outdated Show resolved Hide resolved

turbaszek requested review from feluelle, mik-laj and potiuk December 6, 2019 14:21

potiuk reviewed Dec 7, 2019

View reviewed changes

potiuk approved these changes Dec 7, 2019

View reviewed changes

potiuk added the area:Scheduler Scheduler or dag parsing Issues label Dec 9, 2019

mik-laj reviewed Dec 9, 2019

View reviewed changes

TESTING.rst Outdated Show resolved Hide resolved

mik-laj reviewed Dec 9, 2019

View reviewed changes

airflow/executors/inprocess_executor.py Outdated Show resolved Hide resolved

mik-laj reviewed Dec 9, 2019

View reviewed changes

airflow/executors/inprocess_executor.py Outdated Show resolved Hide resolved

ashb reviewed Dec 9, 2019

View reviewed changes

airflow/models/taskinstance.py Outdated Show resolved Hide resolved

ashb reviewed Dec 9, 2019

View reviewed changes

airflow/executors/executor_loader.py Outdated Show resolved Hide resolved

ashb reviewed Dec 9, 2019

View reviewed changes

airflow/executors/inprocess_executor.py Outdated Show resolved Hide resolved

ashb reviewed Dec 9, 2019

View reviewed changes

kaxil reviewed Dec 10, 2019

View reviewed changes

airflow/models/taskinstance.py Outdated Show resolved Hide resolved

kaxil reviewed Dec 10, 2019

View reviewed changes

docs/executor/debug.rst Outdated Show resolved Hide resolved

turbaszek requested a review from potiuk December 10, 2019 20:32

dimberman reviewed Dec 11, 2019

View reviewed changes

dimberman approved these changes Dec 11, 2019

View reviewed changes

tests/executors/test_inprocess_executor.py Outdated Show resolved Hide resolved

evgenyshulman and others added 7 commits December 12, 2019 15:03

[AIRFLOW-6181] Add InProcessExecutor

51c1367

Adds new executor that is meant to be used mainly for debugging and DAG development purposes. This executor executes single task instance at time and is able to work with SQL Lite and sensors.

fixup! [AIRFLOW-6181] Add InProcessExecutor

31e09b5

fixup! [AIRFLOW-6181] Add InProcessExecutor

e42ed60

fixup! [AIRFLOW-6181] Add InProcessExecutor

df53d02

fixup! [AIRFLOW-6181] Add InProcessExecutor

622cf2d

fixup! [AIRFLOW-6181] Add InProcessExecutor

7d047a5

fixup! [AIRFLOW-6181] Add InProcessExecutor

fbae6ac

turbaszek force-pushed the feature/python_executor branch from e535d94 to fbae6ac Compare December 12, 2019 14:03

mik-laj approved these changes Dec 12, 2019

View reviewed changes

mik-laj merged commit fe2334f into apache:master Dec 12, 2019

potiuk added a commit that referenced this pull request Jan 21, 2020

fixup! [AIRFLOW-6181] Add InProcessExecutor (#6740)

d87efec

potiuk added a commit to potiuk/airflow that referenced this pull request Jan 21, 2020

fixup! [AIRFLOW-6181] Add InProcessExecutor (apache#6740)

102def0

potiuk added a commit to potiuk/airflow that referenced this pull request Jan 22, 2020

fixup! fixup! [AIRFLOW-6181] Add InProcessExecutor (apache#6740)

20ecdeb

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[AIRFLOW-6181] Add InProcessExecutor #6740

[AIRFLOW-6181] Add InProcessExecutor #6740

turbaszek commented Dec 5, 2019 •

edited by dimberman

codecov-io commented Dec 5, 2019 •

edited

KevinYang21 commented Dec 5, 2019

mik-laj commented Dec 5, 2019

turbaszek commented Dec 5, 2019

potiuk commented Dec 7, 2019

potiuk Dec 7, 2019

dimberman Dec 11, 2019

potiuk commented Dec 7, 2019

potiuk commented Dec 9, 2019

potiuk commented Dec 9, 2019

ashb commented Dec 9, 2019

ashb Dec 9, 2019

ashb Dec 9, 2019

turbaszek Dec 9, 2019 •

edited

potiuk commented Dec 9, 2019 •

edited

ashb commented Dec 9, 2019

ashb Dec 9, 2019

potiuk Dec 9, 2019

ashb left a comment

potiuk commented Dec 9, 2019

dimberman Dec 11, 2019

dimberman left a comment

potiuk commented Dec 12, 2019

dimberman commented Dec 12, 2019

	with sensors, sensor's ``mode`` will be automatically set to "reschedule".
	with sensors, all sensors ``mode`` will be automatically set to "reschedule".

[AIRFLOW-6181] Add InProcessExecutor #6740

[AIRFLOW-6181] Add InProcessExecutor #6740

Conversation

turbaszek commented Dec 5, 2019 • edited by dimberman

Jira

Description

Tests

Commits

Documentation

codecov-io commented Dec 5, 2019 • edited

Codecov Report

KevinYang21 commented Dec 5, 2019

mik-laj commented Dec 5, 2019

turbaszek commented Dec 5, 2019

potiuk commented Dec 7, 2019

potiuk Dec 7, 2019

Choose a reason for hiding this comment

dimberman Dec 11, 2019

Choose a reason for hiding this comment

potiuk commented Dec 7, 2019

potiuk commented Dec 9, 2019

potiuk commented Dec 9, 2019

ashb commented Dec 9, 2019

ashb Dec 9, 2019

Choose a reason for hiding this comment

ashb Dec 9, 2019

Choose a reason for hiding this comment

turbaszek Dec 9, 2019 • edited

Choose a reason for hiding this comment

potiuk commented Dec 9, 2019 • edited

ashb commented Dec 9, 2019

ashb Dec 9, 2019

Choose a reason for hiding this comment

potiuk Dec 9, 2019

Choose a reason for hiding this comment

ashb left a comment

Choose a reason for hiding this comment

potiuk commented Dec 9, 2019

dimberman Dec 11, 2019

Choose a reason for hiding this comment

dimberman left a comment

Choose a reason for hiding this comment

potiuk commented Dec 12, 2019

dimberman commented Dec 12, 2019

turbaszek commented Dec 5, 2019 •

edited by dimberman

codecov-io commented Dec 5, 2019 •

edited

turbaszek Dec 9, 2019 •

edited

potiuk commented Dec 9, 2019 •

edited