You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "AutomationDev85 (via GitHub)" <gi...@apache.org> on 2023/04/18 11:40:14 UTC
[GitHub] [airflow] AutomationDev85 opened a new pull request, #30704: Save scheduler execution time by caching dags
AutomationDev85 opened a new pull request, #30704:
URL: https://github.com/apache/airflow/pull/30704
Hi airflow community,
this is my second PR and be happy to work on the scheduler runtime again. We faced an issue with slow scheduler execution time by having millions of queued dag_runs for one DAG.
This PR will add at 2 points in the code a caching of the dag. This saved a lot of scheduler runtime during scheduling many dag_runs for the same dag. The code currently reads the dag out of the DB and if you have a lot of short running tasks this is executed a lot. E.g. we wanted to schedule a DAG with:
max_active_tasks=60,
max_active_runs=180,
and most tasks with an execution time of 2 sec and 1 million in queued state. With the caching we were able to increase scheduler performance. Because the time on our slow DB to query the dag took between 50ms and 250ms and if you execute this only once or 60 times during one scheduler loop run this makes a big change.
@vandonr-amz fyi, as discussed with @jens-scheffler-bosch
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a diff in pull request #30704: Save scheduler execution time by caching dags
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #30704:
URL: https://github.com/apache/airflow/pull/30704#discussion_r1187161127
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1083,8 +1084,11 @@ def _do_scheduling(self, session: Session) -> int:
callback_tuples = self._schedule_all_dag_runs(guard, dag_runs, session)
# Send the callbacks after we commit to ensure the context is up to date when it gets run
+ # cache saves time during scheduling of many dag_runs for same dag
+ cached_get_dag: Callable = lru_cache()(lambda dag_id: self.dagbag.get_dag(dag_id, session=session))
Review Comment:
Could benefit from using `functools.partial` instead. Also itβd be a good idea to improve the type hint here so when the callable is called its return value can also be typed instead of being a useless `Any`.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] jens-scheffler-bosch commented on a diff in pull request #30704: Save scheduler execution time by caching dags
Posted by "jens-scheffler-bosch (via GitHub)" <gi...@apache.org>.
jens-scheffler-bosch commented on code in PR #30704:
URL: https://github.com/apache/airflow/pull/30704#discussion_r1193184497
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1083,8 +1084,11 @@ def _do_scheduling(self, session: Session) -> int:
callback_tuples = self._schedule_all_dag_runs(guard, dag_runs, session)
# Send the callbacks after we commit to ensure the context is up to date when it gets run
+ # cache saves time during scheduling of many dag_runs for same dag
+ cached_get_dag: Callable = lru_cache()(lambda dag_id: self.dagbag.get_dag(dag_id, session=session))
Review Comment:
Thanks for the proposal and thinking a moment about this, it makes the code really "nicer" now. As @AutomationDev85 is off on vacation I'm pushing the update "on behalf".
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1523867035
> > This functionality is basically `functools.lru_cache`.
>
> yes and no because this is an in-function cache. `functools.lru_cache` is intended to provide a cache that is persisted at the object level I think ?
Not realy. @uranusjr is right. lru_cache caches the value of function keyed by arguments of the function (the only condition is that the arguments must be hashable). So it does precisely what the extra (unneded) code does in this case..
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vandonr-amz commented on a diff in pull request #30704: Save scheduler execution time by caching dags
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30704:
URL: https://github.com/apache/airflow/pull/30704#discussion_r1170303898
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1083,8 +1083,13 @@ def _do_scheduling(self, session: Session) -> int:
callback_tuples = self._schedule_all_dag_runs(guard, dag_runs, session)
# Send the callbacks after we commit to ensure the context is up to date when it gets run
+ # cache saves time during scheduling of many dag_runs for same dag
+ cached_dags: dict = {}
for dag_run, callback_to_run in callback_tuples:
- dag = self.dagbag.get_dag(dag_run.dag_id, session=session)
+ if dag_run.dag_id not in cached_dags.keys():
+ cached_dags[dag_run.dag_id] = self.dagbag.get_dag(dag_run.dag_id, session=session)
+
+ dag = cached_dags[dag_run.dag_id]
Review Comment:
nit: when doing this, you do an extra lookup in the dict (when doing `x in dict.keys`) that can be avoided.
Here is how you can rewrite it to minimize dict accesses:
```
dag = cached_dags.get(dag_run.dag_id) # returns None on miss
if dag is None:
dag = self.dagbag.get_dag(dag_run.dag_id, session=session)
cached_dags[dag_run.dag_id] = dag
```
Also, this code is duplicated between the two methods. It'd be nice to wrap this into a `get_or_insert(dict, key, lambda)` function that could go in some utils file maybe ?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] jens-scheffler-bosch commented on a diff in pull request #30704: Save scheduler execution time by caching dags
Posted by "jens-scheffler-bosch (via GitHub)" <gi...@apache.org>.
jens-scheffler-bosch commented on code in PR #30704:
URL: https://github.com/apache/airflow/pull/30704#discussion_r1187920128
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1083,8 +1084,11 @@ def _do_scheduling(self, session: Session) -> int:
callback_tuples = self._schedule_all_dag_runs(guard, dag_runs, session)
# Send the callbacks after we commit to ensure the context is up to date when it gets run
+ # cache saves time during scheduling of many dag_runs for same dag
+ cached_get_dag: Callable = lru_cache()(lambda dag_id: self.dagbag.get_dag(dag_id, session=session))
Review Comment:
I tried to think about how the `functools.partial` could make it better. The proposed statement is a one-liner in adding the `@lru_cache` decorator inline w/o a local additional function declaration. How do you mean it would be better?
In my VSCode at least it shows type hints from parsing python, resulting callable is hinted and the returned `dag` variable in line 1090 is displayed as `(variable) dag: DAG | None` - is it displayed w/o type hints in your IDE?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vandonr-amz commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1524033149
oh nice, I didn't know it could be used like that ! Looks great to me !
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1553752670
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] jens-scheffler-bosch commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "jens-scheffler-bosch (via GitHub)" <gi...@apache.org>.
jens-scheffler-bosch commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1553390742
> Pending CI
CI is done now :-D
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vandonr-amz commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1523937006
> lru_cache caches the value of function keyed by arguments of the function (the only condition is that the arguments must be hashable). So it does precisely what the extra (unneded) code does in this case..
ok, but where is that value cached ?
If the annotated method is global, the cache is global I suppose ? If it's an class method, the cache is stored with the object instance ?
Here we want to cache just within the scope of the function, so we could write an annotated method as a sub-method, but then we have to write it twice, in both locations where it's used ?
We can certainly do that, but I suggested extracting the duplicated code to a function in the first place to avoid this ^^
Or maybe there is a third way I'm missing ? With an lru_cache that is not duplicated and that doesn't cache _too much_ ?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1526494363
BTW. If you really want, you could even annotate it within the non-anonymous lambda (ie. function inside function. Function in Python can easily be declared and run inside another function.
```
class Class:
def method(self, .....):
@lru_cache()...
def cached_call():
# body of cached call
....
# body of object method
....
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1523803577
The added function can basically be replaced by `functools.lru_cache`.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a diff in pull request #30704: Save scheduler execution time by caching dags
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #30704:
URL: https://github.com/apache/airflow/pull/30704#discussion_r1188477820
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1083,8 +1084,11 @@ def _do_scheduling(self, session: Session) -> int:
callback_tuples = self._schedule_all_dag_runs(guard, dag_runs, session)
# Send the callbacks after we commit to ensure the context is up to date when it gets run
+ # cache saves time during scheduling of many dag_runs for same dag
+ cached_get_dag: Callable = lru_cache()(lambda dag_id: self.dagbag.get_dag(dag_id, session=session))
Review Comment:
VS Code uses Pylance, which uses additional guessing to infer types. The heuristics mostly work well and are very helpful, but are not always flawless. Correctly annotating the callable would be more friendly for people using other tools.
Mypy identifies the type to be `Any` (using `typing.reveal_type()` to inspect):
```
$ mypy airflow/jobs/scheduler_job_runner.py
airflow/jobs/scheduler_job_runner.py:1092: note: Revealed type is "Any"
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] jens-scheffler-bosch commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "jens-scheffler-bosch (via GitHub)" <gi...@apache.org>.
jens-scheffler-bosch commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1524026089
> > > This functionality is basically `functools.lru_cache`.
> >
> >
> > yes and no because this is an in-function cache. `functools.lru_cache` is intended to provide a cache that is persisted at the object level I think ?
>
> Not realy. @uranusjr is right. lru_cache caches the value of function keyed by arguments of the function (the only condition is that the arguments must be hashable). So it does precisely what the extra (unneded) code does in this case..
Only being a 95% Python expert (need to earn some stars), removing code with functools sounds good. But also I suppose adding a `@lru_cache` to `self.dagbag.get_dag()` on a general level is not what we want - so to only locally optimize I assume the best approach would be also using a `lambda` like proposed in answer #1 here? https://stackoverflow.com/questions/10270360/python-use-lru-cache-on-lambda-function-or-other-ways-to-create-cache-for-lamb
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vandonr-amz commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1523854954
> This functionality is basically `functools.lru_cache`.
yes and no because this is an in-function cache. `functools.lru_cache` is intended to provide a cache that is persisted at the object level I think ?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk merged pull request #30704: Save scheduler execution time by caching dags
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk merged PR #30704:
URL: https://github.com/apache/airflow/pull/30704
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vandonr-amz commented on a diff in pull request #30704: Save scheduler execution time by caching dags
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30704:
URL: https://github.com/apache/airflow/pull/30704#discussion_r1172974975
##########
airflow/utils/helpers.py:
##########
@@ -45,6 +45,15 @@
S = TypeVar("S")
+def get_value_with_cache(cache: dict, key: str, insert_fn: Callable, *arg, **kwargs) -> Any:
+ """Returns value from cache or function"""
+ return_value = cache.get(key)
+ if not return_value:
+ return_value = cache[key] = insert_fn(*arg, **kwargs)
+
+ return return_value
Review Comment:
ok makes sense then. That specific case doesn't really matter for the current application anyway.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vandonr-amz commented on a diff in pull request #30704: Save scheduler execution time by caching dags
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #30704:
URL: https://github.com/apache/airflow/pull/30704#discussion_r1171808855
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1348,8 +1354,14 @@ def _update_state(dag: DAG, dag_run: DagRun):
tags={"dag_id": dag.dag_id},
)
+ # cache saves time during scheduling of many dag_runs for same dag
+ cached_dags: dict = {}
+
for dag_run in dag_runs:
- dag = dag_run.dag = self.dagbag.get_dag(dag_run.dag_id, session=session)
+ dag = dag_run.dag = get_value_with_cache(
+ cached_dags, dag_run.dag_id, self.dagbag.get_dag, dag_run.dag_id, session=session
+ )
Review Comment:
```suggestion
dag = dag_run.dag = get_value_with_cache(
cached_dags, dag_run.dag_id, lambda: self.dagbag.get_dag(dag_run.dag_id, session=session)
)
```
##########
airflow/utils/helpers.py:
##########
@@ -45,6 +45,15 @@
S = TypeVar("S")
+def get_value_with_cache(cache: dict, key: str, insert_fn: Callable, *arg, **kwargs) -> Any:
+ """Returns value from cache or function"""
+ return_value = cache.get(key)
+ if not return_value:
+ return_value = cache[key] = insert_fn(*arg, **kwargs)
+
+ return return_value
Review Comment:
I'll let you fix the test if you apply this change ;)
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1083,8 +1084,13 @@ def _do_scheduling(self, session: Session) -> int:
callback_tuples = self._schedule_all_dag_runs(guard, dag_runs, session)
# Send the callbacks after we commit to ensure the context is up to date when it gets run
+ # cache saves time during scheduling of many dag_runs for same dag
+ cached_dags: dict = {}
for dag_run, callback_to_run in callback_tuples:
- dag = self.dagbag.get_dag(dag_run.dag_id, session=session)
+ dag = get_value_with_cache(
+ cached_dags, dag_run.dag_id, self.dagbag.get_dag, dag_run.dag_id, session=session
+ )
Review Comment:
Passing the params individually is a convoluted way to do it... Using a lambda you can just pass the whole packaged call as a Callable without the need for extra parameters
```suggestion
dag = get_value_with_cache(
cached_dags, dag_run.dag_id, lambda: self.dagbag.get_dag(dag_run.dag_id, session=session)
)
```
##########
airflow/utils/helpers.py:
##########
@@ -45,6 +45,15 @@
S = TypeVar("S")
+def get_value_with_cache(cache: dict, key: str, insert_fn: Callable, *arg, **kwargs) -> Any:
+ """Returns value from cache or function"""
+ return_value = cache.get(key)
+ if not return_value:
+ return_value = cache[key] = insert_fn(*arg, **kwargs)
+
+ return return_value
Review Comment:
By using a lambda, we don't need to handle paramters here. (suggesting a rename of the function passed as well ? WDYT ?)
Also, `if not return_value` will trigger if the saved value is "falsy", which is the case for None, but also for empty lists/dict for instance, which can lead to surprising behavior. I think explicitly checking against None would be better here.
```suggestion
def get_value_with_cache(cache: dict, key: str, do_get_value: Callable) -> Any:
"""Returns value from cache or function"""
return_value = cache.get(key)
if return_value is None:
return_value = cache[key] = do_get_value()
return return_value
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1530260877
Second pair of π and π needed :)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1528890874
Second pair of eyes needed here , as this is core part.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AutomationDev85 commented on a diff in pull request #30704: Save scheduler execution time by caching dags
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on code in PR #30704:
URL: https://github.com/apache/airflow/pull/30704#discussion_r1172229514
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1083,8 +1084,13 @@ def _do_scheduling(self, session: Session) -> int:
callback_tuples = self._schedule_all_dag_runs(guard, dag_runs, session)
# Send the callbacks after we commit to ensure the context is up to date when it gets run
+ # cache saves time during scheduling of many dag_runs for same dag
+ cached_dags: dict = {}
for dag_run, callback_to_run in callback_tuples:
- dag = self.dagbag.get_dag(dag_run.dag_id, session=session)
+ dag = get_value_with_cache(
+ cached_dags, dag_run.dag_id, self.dagbag.get_dag, dag_run.dag_id, session=session
+ )
Review Comment:
Thanks for the feedback! Using now lambda.
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1348,8 +1354,14 @@ def _update_state(dag: DAG, dag_run: DagRun):
tags={"dag_id": dag.dag_id},
)
+ # cache saves time during scheduling of many dag_runs for same dag
+ cached_dags: dict = {}
+
for dag_run in dag_runs:
- dag = dag_run.dag = self.dagbag.get_dag(dag_run.dag_id, session=session)
+ dag = dag_run.dag = get_value_with_cache(
+ cached_dags, dag_run.dag_id, self.dagbag.get_dag, dag_run.dag_id, session=session
+ )
Review Comment:
Thanks for the feedback! Using now lambda.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AutomationDev85 commented on a diff in pull request #30704: Save scheduler execution time by caching dags
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on code in PR #30704:
URL: https://github.com/apache/airflow/pull/30704#discussion_r1172235531
##########
airflow/utils/helpers.py:
##########
@@ -45,6 +45,15 @@
S = TypeVar("S")
+def get_value_with_cache(cache: dict, key: str, insert_fn: Callable, *arg, **kwargs) -> Any:
+ """Returns value from cache or function"""
+ return_value = cache.get(key)
+ if not return_value:
+ return_value = cache[key] = insert_fn(*arg, **kwargs)
+
+ return return_value
Review Comment:
Thanks for the feedback. I like the new name for the function!
I'm using `if not return_value` because if the cache returns empty value, I think the best way is to call do_get_value to make sure that the right and latest value is used and not the cached value.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Save scheduler execution time by caching dags [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1783877399
@uranusjr @jens-scheffler-bosch @potiuk This PR makes no sense btw -- DagBag.get_dag _already_ has a cache:
https://github.com/apache/airflow/blob/d1c58d86de1267d9268a1efe0a0c102633c051a1/airflow/models/dagbag.py#L189-L192
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1547104001
Pending CI
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AutomationDev85 commented on a diff in pull request #30704: Save scheduler execution time by caching dags
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on code in PR #30704:
URL: https://github.com/apache/airflow/pull/30704#discussion_r1171453144
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1083,8 +1083,13 @@ def _do_scheduling(self, session: Session) -> int:
callback_tuples = self._schedule_all_dag_runs(guard, dag_runs, session)
# Send the callbacks after we commit to ensure the context is up to date when it gets run
+ # cache saves time during scheduling of many dag_runs for same dag
+ cached_dags: dict = {}
for dag_run, callback_to_run in callback_tuples:
- dag = self.dagbag.get_dag(dag_run.dag_id, session=session)
+ if dag_run.dag_id not in cached_dags.keys():
+ cached_dags[dag_run.dag_id] = self.dagbag.get_dag(dag_run.dag_id, session=session)
+
+ dag = cached_dags[dag_run.dag_id]
Review Comment:
Thanks for your feedback! This is a good idea, i added a function to the helpers, not sure is this is the right place but looking forward for your feedback :)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1526472254
> Or maybe there is a third way I'm missing ? With an lru_cache that is not duplicated and that doesn't cache too much ?
1. Make a class methods wrapping the global method call
2. Annotate it with @lru_cache
Saves about 80% of the code.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vandonr-amz commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1526661059
I like the lambda solution, I think it looks cleaner than having a function inside the function
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #30704: Save scheduler execution time by caching dags
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #30704:
URL: https://github.com/apache/airflow/pull/30704#issuecomment-1526699838
> I like the lambda solution, I think it looks cleaner than having a function inside the function
Yes it has an appeal. the most important is that we are building of lru_cache, rather than trying to do what it does
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org