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/10/07 09:52:41 UTC
[GitHub] [airflow] ashb opened a new pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
ashb opened a new pull request #11327:
URL: https://github.com/apache/airflow/pull/11327
Spawning a whole new python process and then re-loading all of Airflow
is expensive. All though this time fades to insignificance for long
running tasks, this delay gives a "bad" experience for new users when
they are just trying out Airflow for the first time.
For the LocalExecutor this cuts the "queued time" down from 1.5s to 0.1s
on average.
Data on this for a simple 10-task sequential DAG:
```sql
SELECT execution_date,
min(start_date - queued_dttm) AS min_quued_delay,
max(start_date - queued_dttm) AS max_queued_delay
FROM task_instance
WHERE dag_id = 'scenario1_case2_03_1' GROUP BY execution_date;
```
| execution_date | min_quued_delay | max_queued_delay | with change? |
| ------------------------------- | ----------------- | ------------------ | -- |
| 2020-10-07 09:24:58.915889+01 | 00:00:01.494198 | 00:00:01.652728 | No |
| 2020-10-06 01:00:00+01 | 00:00:00.08397 | 00:00:00.10932 | Yes |
This was discovered in my general benchmarking and profiling of the scheduler for AIP-15, but it's not tied to any of that work. There are more of these kind of improvements coming, each unrelated but all add up.
<!--
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] ashb commented on pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#issuecomment-705553009
Hmmm, something broken in all the kube test unrelated to this.
```
./scripts/ci/kubernetes/ci_run_kubernetes_tests.sh: /home/runner/work/airflow/airflow/.build/.kubernetes_venv/bin/pip: /home/runner/work/airflow/airflow/.build/.kubernetes_venv/bin/python: bad interpreter: No such file or directory
```
----------------------------------------------------------------
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] kaxil commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r501016972
##########
File path: airflow/config_templates/default_airflow.cfg
##########
@@ -133,6 +133,11 @@ load_default_connections = True
# Where your Airflow plugins are stored
plugins_folder = {AIRFLOW_HOME}/plugins
+# Should tasks be executed via forking of the parent process ("False",
+# the speedier option) or by spawning a new python process ("True" slow,
+# but means plugin changes picked up by tasks straight away)
+# Default: execute_tasks_new_python_interpreter = False
+
Review comment:
This looks a bit odd, why not just have it as:
```ini
execute_tasks_new_python_interpreter = False
```
?
----------------------------------------------------------------
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] ashb commented on pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#issuecomment-704932572
----------------------------------------------------------------
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] ashb commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r500909630
##########
File path: airflow/executors/local_executor.py
##########
@@ -65,14 +74,31 @@ def execute_work(self, key: TaskInstanceKey, command: CommandType) -> None:
if key is None:
return
self.log.info("%s running %s", self.__class__.__name__, command)
+ old_title = getproctitle()
try:
- subprocess.check_call(command, close_fds=True)
+ from airflow.cli.cli_parser import get_parser
+ parser = get_parser()
+ # [1:] - remove "airflow" from the start of the command
+ args = parser.parse_args(command[1:])
Review comment:
My point is that custom operators don't need to be in plugins - just a normal python module, and then they won't be loaded by these long running process. See #9498 and #9500.
This concern still applies to hooks though, so I'll add a config flag
----------------------------------------------------------------
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] ashb merged pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb merged pull request #11327:
URL: https://github.com/apache/airflow/pull/11327
----------------------------------------------------------------
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] ashb merged pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb merged pull request #11327:
URL: https://github.com/apache/airflow/pull/11327
----------------------------------------------------------------
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] ashb commented on pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#issuecomment-705449207
----------------------------------------------------------------
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] ashb commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r501011974
##########
File path: airflow/executors/local_executor.py
##########
@@ -65,14 +74,31 @@ def execute_work(self, key: TaskInstanceKey, command: CommandType) -> None:
if key is None:
return
self.log.info("%s running %s", self.__class__.__name__, command)
+ old_title = getproctitle()
try:
- subprocess.check_call(command, close_fds=True)
+ from airflow.cli.cli_parser import get_parser
+ parser = get_parser()
+ # [1:] - remove "airflow" from the start of the command
+ args = parser.parse_args(command[1:])
Review comment:
Done.
----------------------------------------------------------------
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] kaxil commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r500892060
##########
File path: airflow/executors/local_executor.py
##########
@@ -65,14 +74,31 @@ def execute_work(self, key: TaskInstanceKey, command: CommandType) -> None:
if key is None:
return
self.log.info("%s running %s", self.__class__.__name__, command)
+ old_title = getproctitle()
try:
- subprocess.check_call(command, close_fds=True)
+ from airflow.cli.cli_parser import get_parser
+ parser = get_parser()
+ # [1:] - remove "airflow" from the start of the command
+ args = parser.parse_args(command[1:])
+
+ setproctitle(f"airflow task supervisor: {command}")
+
+ args.func(args)
state = State.SUCCESS
- except subprocess.CalledProcessError as e:
+ except BaseException as e:
state = State.FAILED
self.log.error("Failed to execute task %s.", str(e))
+ finally:
+ setproctitle(old_title)
self.result_queue.put((key, state))
+ @abstractmethod
+ def do_work(self):
+ """
+ Called in the subprocess and should get then execute tasks
Review comment:
```suggestion
Called in the subprocess and should then execute tasks
```
----------------------------------------------------------------
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] ashb commented on pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#issuecomment-705666308
Oh, those kube tests will fail because I haven't re-triggered the image build. If the rests of the tests pass I'll merge this.
----------------------------------------------------------------
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] ashb commented on pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#issuecomment-705139313
I just noticed that we still need a fork in here - `task_run` in cli.commands does a `logging.shutdown` (and we need that as it does the remote upload) so I'll change this to fork first.
----------------------------------------------------------------
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] ashb commented on pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#issuecomment-705449207
Updated benchmarks -- the fork is basically free.
----------------------------------------------------------------
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] ashb commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r500909630
##########
File path: airflow/executors/local_executor.py
##########
@@ -65,14 +74,31 @@ def execute_work(self, key: TaskInstanceKey, command: CommandType) -> None:
if key is None:
return
self.log.info("%s running %s", self.__class__.__name__, command)
+ old_title = getproctitle()
try:
- subprocess.check_call(command, close_fds=True)
+ from airflow.cli.cli_parser import get_parser
+ parser = get_parser()
+ # [1:] - remove "airflow" from the start of the command
+ args = parser.parse_args(command[1:])
Review comment:
My point is that custom operators don't need to be in plugins - just a normal python module, and then they won't be loaded by these long running process. See #9498 and #9500.
This concern still applies to hooks though.
----------------------------------------------------------------
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] kaxil commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r501019063
##########
File path: airflow/config_templates/default_airflow.cfg
##########
@@ -133,6 +133,11 @@ load_default_connections = True
# Where your Airflow plugins are stored
plugins_folder = {AIRFLOW_HOME}/plugins
+# Should tasks be executed via forking of the parent process ("False",
+# the speedier option) or by spawning a new python process ("True" slow,
+# but means plugin changes picked up by tasks straight away)
+# Default: execute_tasks_new_python_interpreter = False
+
Review comment:
Or atleast:
```
# execute_tasks_new_python_interpreter = False
```
The `Default:` in the front of `key = val` might make users search for the config below the comments, at least a the first glance.
----------------------------------------------------------------
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 #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r500890476
##########
File path: airflow/executors/local_executor.py
##########
@@ -65,14 +74,31 @@ def execute_work(self, key: TaskInstanceKey, command: CommandType) -> None:
if key is None:
return
self.log.info("%s running %s", self.__class__.__name__, command)
+ old_title = getproctitle()
try:
- subprocess.check_call(command, close_fds=True)
+ from airflow.cli.cli_parser import get_parser
+ parser = get_parser()
+ # [1:] - remove "airflow" from the start of the command
+ args = parser.parse_args(command[1:])
Review comment:
I think this may raise similar concerns as #6905: that users will have to restart Airflow scheduler to reload plugins, right?
----------------------------------------------------------------
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 #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r500905845
##########
File path: airflow/executors/local_executor.py
##########
@@ -65,14 +74,31 @@ def execute_work(self, key: TaskInstanceKey, command: CommandType) -> None:
if key is None:
return
self.log.info("%s running %s", self.__class__.__name__, command)
+ old_title = getproctitle()
try:
- subprocess.check_call(command, close_fds=True)
+ from airflow.cli.cli_parser import get_parser
+ parser = get_parser()
+ # [1:] - remove "airflow" from the start of the command
+ args = parser.parse_args(command[1:])
Review comment:
DAGs in plugins are one thing, custom operators are other and I think many users keep custom operators as plugins
----------------------------------------------------------------
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] kaxil commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r501294227
##########
File path: airflow/settings.py
##########
@@ -365,3 +365,11 @@ def initialize():
# to get all the logs from the print & log statements in the DAG files before a task is run
# The handlers are restored after the task completes execution.
DONOT_MODIFY_HANDLERS = conf.getboolean('logging', 'donot_modify_handlers', fallback=False)
+
+CAN_FORK = hasattr(os, "fork")
+
+EXECUTE_TASKS_NEW_PYTHON_INTERPRETER = CAN_FORK and conf.getboolean(
Review comment:
```suggestion
EXECUTE_TASKS_NEW_PYTHON_INTERPRETER = not CAN_FORK or conf.getboolean(
```
----------------------------------------------------------------
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] ashb commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r501314811
##########
File path: airflow/settings.py
##########
@@ -365,3 +365,11 @@ def initialize():
# to get all the logs from the print & log statements in the DAG files before a task is run
# The handlers are restored after the task completes execution.
DONOT_MODIFY_HANDLERS = conf.getboolean('logging', 'donot_modify_handlers', fallback=False)
+
+CAN_FORK = hasattr(os, "fork")
+
+EXECUTE_TASKS_NEW_PYTHON_INTERPRETER = CAN_FORK and conf.getboolean(
Review comment:
yup.
----------------------------------------------------------------
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] ashb commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r500895612
##########
File path: airflow/executors/local_executor.py
##########
@@ -65,14 +74,31 @@ def execute_work(self, key: TaskInstanceKey, command: CommandType) -> None:
if key is None:
return
self.log.info("%s running %s", self.__class__.__name__, command)
+ old_title = getproctitle()
try:
- subprocess.check_call(command, close_fds=True)
+ from airflow.cli.cli_parser import get_parser
+ parser = get_parser()
+ # [1:] - remove "airflow" from the start of the command
+ args = parser.parse_args(command[1:])
Review comment:
Oh I hadn't seen that PR (or forgotten about it). Forgot since I commented on it!
Hmm, not sure - yes plugins would be pre-loaded. I guess this Congress down to not recommending putting operators in plugins too.
I'll add a config flag (on by default?) to go back to the old way
----------------------------------------------------------------
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] ashb commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r501027480
##########
File path: airflow/config_templates/default_airflow.cfg
##########
@@ -133,6 +133,11 @@ load_default_connections = True
# Where your Airflow plugins are stored
plugins_folder = {AIRFLOW_HOME}/plugins
+# Should tasks be executed via forking of the parent process ("False",
+# the speedier option) or by spawning a new python process ("True" slow,
+# but means plugin changes picked up by tasks straight away)
+# Default: execute_tasks_new_python_interpreter = False
+
Review comment:
I guess I shouldn't start a new convention here, but this was driven by the sort of change in https://github.com/apache/airflow/pull/11322. I'll pull this out.
----------------------------------------------------------------
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] ashb commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r501292179
##########
File path: airflow/settings.py
##########
@@ -365,3 +365,11 @@ def initialize():
# to get all the logs from the print & log statements in the DAG files before a task is run
# The handlers are restored after the task completes execution.
DONOT_MODIFY_HANDLERS = conf.getboolean('logging', 'donot_modify_handlers', fallback=False)
+
+CAN_FORK = hasattr(os, "fork")
+
+EXECUTE_TASKS_NEW_PYTHON_INTERPRETER = CAN_FORK and conf.getboolean(
Review comment:
This is wrong too.
----------------------------------------------------------------
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] ashb commented on pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#issuecomment-704936366
I was thinking I'd follow up with a separate PR to add the same to CeleryExecutor.
----------------------------------------------------------------
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] ashb commented on a change in pull request #11327: Reduce "start-up" time for tasks in LocalExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #11327:
URL: https://github.com/apache/airflow/pull/11327#discussion_r500895612
##########
File path: airflow/executors/local_executor.py
##########
@@ -65,14 +74,31 @@ def execute_work(self, key: TaskInstanceKey, command: CommandType) -> None:
if key is None:
return
self.log.info("%s running %s", self.__class__.__name__, command)
+ old_title = getproctitle()
try:
- subprocess.check_call(command, close_fds=True)
+ from airflow.cli.cli_parser import get_parser
+ parser = get_parser()
+ # [1:] - remove "airflow" from the start of the command
+ args = parser.parse_args(command[1:])
Review comment:
Oh I hadn't seen that PR (or forgotten about it).
Hmm, not sure - yes plugins would be pre-loaded. I guess this Congress down to not recommending putting operators in plugins too.
I'll add a config flag (on by default?) to go back to the old way
----------------------------------------------------------------
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