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 2022/06/07 12:14:22 UTC
[GitHub] [airflow] ashb opened a new pull request, #24284: Speed up grid_data endpoint by 10x
ashb opened a new pull request, #24284:
URL: https://github.com/apache/airflow/pull/24284
These changes make the endpoint go from almost 20s down to 1.5s and the
changes are two fold:
1. Keep datetimes as objects for as long as possible
Previously we were converting start/end dates for a task group to a
string, and then in the parent parsing it back to a datetime to find
the min and max of all the child nodes.
The fix for that was to leave it as a datetime (or a
pendulum.DateTime technically) and use the existing
`AirflowJsonEncoder` class to "correctly" encode these objects on
output.
2. Reduce the number of DB queries from 1 per task to 1.
The removed `get_task_summaries` function was called for each task,
and was making a query to the database to find info for the given
DagRuns.
The helper function now makes just a single DB query for all
tasks/runs and constructs a dict to efficiently look up the ti by
run_id.
--
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] ashb commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891572059
##########
airflow/www/views.py:
##########
@@ -250,62 +251,102 @@ def _safe_parse_datetime(v):
abort(400, f"Invalid datetime: {v!r}")
-def task_group_to_grid(task_item_or_group, dag, dag_runs, session):
+def dag_to_grid(dag, dag_runs, session):
"""
- Create a nested dict representation of this TaskGroup and its children used to construct
- the Graph.
+ Create a nested dict representation of the DAG's TaskGroup and its children
+ used to construct the Graph and Grid views.
"""
- if isinstance(task_item_or_group, AbstractOperator):
- return {
- 'id': task_item_or_group.task_id,
- 'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
- 'label': task_item_or_group.label,
- 'extra_links': task_item_or_group.extra_links,
- 'is_mapped': task_item_or_group.is_mapped,
- }
+ query = (
+ session.query(
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.state,
+ TaskInstance.start_date,
+ TaskInstance.end_date,
+ TaskInstance._try_number,
+ )
+ .filter(
+ TaskInstance.dag_id == dag.dag_id,
+ TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
+ # Only get normal task instances or the first mapped task
+ TaskInstance.map_index <= 0,
+ )
+ .order_by(TaskInstance.task_id)
+ )
- # Task Group
- task_group = task_item_or_group
+ grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
- children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
+ def task_group_to_grid(item, dag_runs, grouped_tis):
+ if isinstance(item, AbstractOperator):
- def get_summary(dag_run, children):
- child_instances = [child['instances'] for child in children if 'instances' in child]
- child_instances = [item for sublist in child_instances for item in sublist]
+ def _get_summary(task_instance):
+ try_count = (
+ task_instance._try_number
+ if task_instance._try_number != 0 or task_instance.state in State.running
+ else task_instance._try_number + 1
+ )
- children_start_dates = [item['start_date'] for item in child_instances if item]
- children_end_dates = [item['end_date'] for item in child_instances if item]
- children_states = [item['state'] for item in child_instances if item]
+ return {
+ 'task_id': task_instance.task_id,
+ 'run_id': task_instance.run_id,
+ 'map_index': task_instance.map_index,
+ 'state': task_instance.state,
+ 'start_date': task_instance.start_date,
+ 'end_date': task_instance.end_date,
+ 'try_number': try_count,
Review Comment:
Do you even need the counts, or can we make the decision of what state to show on the server side?
--
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] ashb commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891578948
##########
airflow/www/views.py:
##########
@@ -250,62 +251,102 @@ def _safe_parse_datetime(v):
abort(400, f"Invalid datetime: {v!r}")
-def task_group_to_grid(task_item_or_group, dag, dag_runs, session):
+def dag_to_grid(dag, dag_runs, session):
"""
- Create a nested dict representation of this TaskGroup and its children used to construct
- the Graph.
+ Create a nested dict representation of the DAG's TaskGroup and its children
+ used to construct the Graph and Grid views.
"""
- if isinstance(task_item_or_group, AbstractOperator):
- return {
- 'id': task_item_or_group.task_id,
- 'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
- 'label': task_item_or_group.label,
- 'extra_links': task_item_or_group.extra_links,
- 'is_mapped': task_item_or_group.is_mapped,
- }
+ query = (
+ session.query(
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.state,
+ TaskInstance.start_date,
+ TaskInstance.end_date,
+ TaskInstance._try_number,
+ )
+ .filter(
+ TaskInstance.dag_id == dag.dag_id,
+ TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
+ # Only get normal task instances or the first mapped task
+ TaskInstance.map_index <= 0,
+ )
+ .order_by(TaskInstance.task_id)
+ )
- # Task Group
- task_group = task_item_or_group
+ grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
- children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
+ def task_group_to_grid(item, dag_runs, grouped_tis):
+ if isinstance(item, AbstractOperator):
- def get_summary(dag_run, children):
- child_instances = [child['instances'] for child in children if 'instances' in child]
- child_instances = [item for sublist in child_instances for item in sublist]
+ def _get_summary(task_instance):
+ try_count = (
+ task_instance._try_number
+ if task_instance._try_number != 0 or task_instance.state in State.running
+ else task_instance._try_number + 1
+ )
- children_start_dates = [item['start_date'] for item in child_instances if item]
- children_end_dates = [item['end_date'] for item in child_instances if item]
- children_states = [item['state'] for item in child_instances if item]
+ return {
+ 'task_id': task_instance.task_id,
+ 'run_id': task_instance.run_id,
+ 'map_index': task_instance.map_index,
+ 'state': task_instance.state,
+ 'start_date': task_instance.start_date,
+ 'end_date': task_instance.end_date,
+ 'try_number': try_count,
Review Comment:
Oh showing the breakdown in the tooltip? If so I'll leave the counts back
--
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] github-actions[bot] commented on pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #24284:
URL: https://github.com/apache/airflow/pull/24284#issuecomment-1148629895
The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.
--
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] ashb merged pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb merged PR #24284:
URL: https://github.com/apache/airflow/pull/24284
--
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] ashb commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891497728
##########
airflow/utils/json.py:
##########
@@ -45,7 +45,7 @@ def __init__(self, *args, **kwargs):
def _default(obj):
"""Convert dates and numpy objects in a json serializable format."""
if isinstance(obj, datetime):
- return obj.strftime('%Y-%m-%dT%H:%M:%SZ')
+ return obj.isoformat()
Review Comment:
`Z` vs `+00:00` shouldn't matter to any compliant parser. So this should be okay
--
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] ashb commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r892530498
##########
airflow/utils/json.py:
##########
@@ -45,7 +45,7 @@ def __init__(self, *args, **kwargs):
def _default(obj):
"""Convert dates and numpy objects in a json serializable format."""
if isinstance(obj, datetime):
- return obj.strftime('%Y-%m-%dT%H:%M:%SZ')
+ return obj.isoformat()
Review Comment:
Oh right -- it shouldn't.
--
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] bbovenzi commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
bbovenzi commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891503654
##########
airflow/www/views.py:
##########
@@ -250,62 +251,102 @@ def _safe_parse_datetime(v):
abort(400, f"Invalid datetime: {v!r}")
-def task_group_to_grid(task_item_or_group, dag, dag_runs, session):
+def dag_to_grid(dag, dag_runs, session):
"""
- Create a nested dict representation of this TaskGroup and its children used to construct
- the Graph.
+ Create a nested dict representation of the DAG's TaskGroup and its children
+ used to construct the Graph and Grid views.
"""
- if isinstance(task_item_or_group, AbstractOperator):
- return {
- 'id': task_item_or_group.task_id,
- 'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
- 'label': task_item_or_group.label,
- 'extra_links': task_item_or_group.extra_links,
- 'is_mapped': task_item_or_group.is_mapped,
- }
+ query = (
+ session.query(
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.state,
+ TaskInstance.start_date,
+ TaskInstance.end_date,
+ TaskInstance._try_number,
+ )
+ .filter(
+ TaskInstance.dag_id == dag.dag_id,
+ TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
+ # Only get normal task instances or the first mapped task
+ TaskInstance.map_index <= 0,
+ )
+ .order_by(TaskInstance.task_id)
+ )
- # Task Group
- task_group = task_item_or_group
+ grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
- children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
+ def task_group_to_grid(item, dag_runs, grouped_tis):
+ if isinstance(item, AbstractOperator):
- def get_summary(dag_run, children):
- child_instances = [child['instances'] for child in children if 'instances' in child]
- child_instances = [item for sublist in child_instances for item in sublist]
+ def _get_summary(task_instance):
+ try_count = (
+ task_instance._try_number
+ if task_instance._try_number != 0 or task_instance.state in State.running
+ else task_instance._try_number + 1
+ )
- children_start_dates = [item['start_date'] for item in child_instances if item]
- children_end_dates = [item['end_date'] for item in child_instances if item]
- children_states = [item['state'] for item in child_instances if item]
+ return {
+ 'task_id': task_instance.task_id,
+ 'run_id': task_instance.run_id,
+ 'map_index': task_instance.map_index,
+ 'state': task_instance.state,
+ 'start_date': task_instance.start_date,
+ 'end_date': task_instance.end_date,
+ 'try_number': try_count,
Review Comment:
Everything else seems to be fine we just need something like `mapped_states`, best format would be like so (I'll make any UI changes needed):
```
{
success: 3.
running: 2,
queued: 1,
}
```
--
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] ashb commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891498551
##########
airflow/www/views.py:
##########
@@ -250,62 +251,102 @@ def _safe_parse_datetime(v):
abort(400, f"Invalid datetime: {v!r}")
-def task_group_to_grid(task_item_or_group, dag, dag_runs, session):
+def dag_to_grid(dag, dag_runs, session):
"""
- Create a nested dict representation of this TaskGroup and its children used to construct
- the Graph.
+ Create a nested dict representation of the DAG's TaskGroup and its children
+ used to construct the Graph and Grid views.
"""
- if isinstance(task_item_or_group, AbstractOperator):
- return {
- 'id': task_item_or_group.task_id,
- 'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
- 'label': task_item_or_group.label,
- 'extra_links': task_item_or_group.extra_links,
- 'is_mapped': task_item_or_group.is_mapped,
- }
+ query = (
+ session.query(
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.state,
+ TaskInstance.start_date,
+ TaskInstance.end_date,
+ TaskInstance._try_number,
+ )
+ .filter(
+ TaskInstance.dag_id == dag.dag_id,
+ TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
+ # Only get normal task instances or the first mapped task
+ TaskInstance.map_index <= 0,
+ )
+ .order_by(TaskInstance.task_id)
+ )
- # Task Group
- task_group = task_item_or_group
+ grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
- children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
+ def task_group_to_grid(item, dag_runs, grouped_tis):
+ if isinstance(item, AbstractOperator):
- def get_summary(dag_run, children):
- child_instances = [child['instances'] for child in children if 'instances' in child]
- child_instances = [item for sublist in child_instances for item in sublist]
+ def _get_summary(task_instance):
+ try_count = (
+ task_instance._try_number
+ if task_instance._try_number != 0 or task_instance.state in State.running
+ else task_instance._try_number + 1
+ )
- children_start_dates = [item['start_date'] for item in child_instances if item]
- children_end_dates = [item['end_date'] for item in child_instances if item]
- children_states = [item['state'] for item in child_instances if item]
+ return {
+ 'task_id': task_instance.task_id,
+ 'run_id': task_instance.run_id,
+ 'map_index': task_instance.map_index,
+ 'state': task_instance.state,
+ 'start_date': task_instance.start_date,
+ 'end_date': task_instance.end_date,
+ 'try_number': try_count,
Review Comment:
Oh yes. Removing that was not intentional!
--
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] ashb commented on pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #24284:
URL: https://github.com/apache/airflow/pull/24284#issuecomment-1154195203
Stack trace for that?
I didn't test with running DAGs. Oops! (Good thing we can add tests now)
--
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] ashb commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891499745
##########
airflow/www/views.py:
##########
@@ -250,62 +251,102 @@ def _safe_parse_datetime(v):
abort(400, f"Invalid datetime: {v!r}")
-def task_group_to_grid(task_item_or_group, dag, dag_runs, session):
+def dag_to_grid(dag, dag_runs, session):
"""
- Create a nested dict representation of this TaskGroup and its children used to construct
- the Graph.
+ Create a nested dict representation of the DAG's TaskGroup and its children
+ used to construct the Graph and Grid views.
"""
- if isinstance(task_item_or_group, AbstractOperator):
- return {
- 'id': task_item_or_group.task_id,
- 'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
- 'label': task_item_or_group.label,
- 'extra_links': task_item_or_group.extra_links,
- 'is_mapped': task_item_or_group.is_mapped,
- }
+ query = (
+ session.query(
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.state,
+ TaskInstance.start_date,
+ TaskInstance.end_date,
+ TaskInstance._try_number,
+ )
+ .filter(
+ TaskInstance.dag_id == dag.dag_id,
+ TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
+ # Only get normal task instances or the first mapped task
+ TaskInstance.map_index <= 0,
+ )
+ .order_by(TaskInstance.task_id)
+ )
- # Task Group
- task_group = task_item_or_group
+ grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
- children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
+ def task_group_to_grid(item, dag_runs, grouped_tis):
+ if isinstance(item, AbstractOperator):
- def get_summary(dag_run, children):
- child_instances = [child['instances'] for child in children if 'instances' in child]
- child_instances = [item for sublist in child_instances for item in sublist]
+ def _get_summary(task_instance):
+ try_count = (
+ task_instance._try_number
+ if task_instance._try_number != 0 or task_instance.state in State.running
+ else task_instance._try_number + 1
+ )
- children_start_dates = [item['start_date'] for item in child_instances if item]
- children_end_dates = [item['end_date'] for item in child_instances if item]
- children_states = [item['state'] for item in child_instances if item]
+ return {
+ 'task_id': task_instance.task_id,
+ 'run_id': task_instance.run_id,
+ 'map_index': task_instance.map_index,
+ 'state': task_instance.state,
+ 'start_date': task_instance.start_date,
+ 'end_date': task_instance.end_date,
+ 'try_number': try_count,
Review Comment:
Where did we include that before? I'm not seeing where I delete it in this diff? (My brain is foggy 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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891498300
##########
airflow/utils/json.py:
##########
@@ -45,7 +45,7 @@ def __init__(self, *args, **kwargs):
def _default(obj):
"""Convert dates and numpy objects in a json serializable format."""
if isinstance(obj, datetime):
- return obj.strftime('%Y-%m-%dT%H:%M:%SZ')
+ return obj.isoformat()
Review Comment:
I've just changed this (to fix tests) so that it uses the existing `is_naive` and `convert_to_utc` functions so it will always include a TZ.
--
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] ashb commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r895947581
##########
airflow/www/views.py:
##########
@@ -250,62 +251,102 @@ def _safe_parse_datetime(v):
abort(400, f"Invalid datetime: {v!r}")
-def task_group_to_grid(task_item_or_group, dag, dag_runs, session):
+def dag_to_grid(dag, dag_runs, session):
"""
- Create a nested dict representation of this TaskGroup and its children used to construct
- the Graph.
+ Create a nested dict representation of the DAG's TaskGroup and its children
+ used to construct the Graph and Grid views.
"""
- if isinstance(task_item_or_group, AbstractOperator):
- return {
- 'id': task_item_or_group.task_id,
- 'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
- 'label': task_item_or_group.label,
- 'extra_links': task_item_or_group.extra_links,
- 'is_mapped': task_item_or_group.is_mapped,
- }
+ query = (
+ session.query(
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.state,
+ TaskInstance.start_date,
+ TaskInstance.end_date,
+ TaskInstance._try_number,
+ )
+ .filter(
+ TaskInstance.dag_id == dag.dag_id,
+ TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
+ # Only get normal task instances or the first mapped task
+ TaskInstance.map_index <= 0,
+ )
+ .order_by(TaskInstance.task_id)
+ )
- # Task Group
- task_group = task_item_or_group
+ grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
- children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
+ def task_group_to_grid(item, dag_runs, grouped_tis):
+ if isinstance(item, AbstractOperator):
- def get_summary(dag_run, children):
- child_instances = [child['instances'] for child in children if 'instances' in child]
- child_instances = [item for sublist in child_instances for item in sublist]
+ def _get_summary(task_instance):
+ try_count = (
+ task_instance._try_number
+ if task_instance._try_number != 0 or task_instance.state in State.running
+ else task_instance._try_number + 1
+ )
- children_start_dates = [item['start_date'] for item in child_instances if item]
- children_end_dates = [item['end_date'] for item in child_instances if item]
- children_states = [item['state'] for item in child_instances if item]
+ return {
+ 'task_id': task_instance.task_id,
+ 'run_id': task_instance.run_id,
+ 'map_index': task_instance.map_index,
+ 'state': task_instance.state,
+ 'start_date': task_instance.start_date,
+ 'end_date': task_instance.end_date,
+ 'try_number': try_count,
Review Comment:
Right, made this change, and updated the React to match in the two places I found. Did I miss anywhere @bbovenzi
--
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 #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #24284:
URL: https://github.com/apache/airflow/pull/24284#issuecomment-1148601116
I guess just transferring and parsing the resulting json will take most of the time now. BTW. Are we using deflate there :D? AnNd there are quite a few tricks with Javascript VM optimization for JSON which we MIGHT take a look at :)
--
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] ashb commented on pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #24284:
URL: https://github.com/apache/airflow/pull/24284#issuecomment-1148609967
There are many much much slower points to optomize on the front end before we have to worry about time spent parsing date times on the client side.
--
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] bbovenzi commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
bbovenzi commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891503654
##########
airflow/www/views.py:
##########
@@ -250,62 +251,102 @@ def _safe_parse_datetime(v):
abort(400, f"Invalid datetime: {v!r}")
-def task_group_to_grid(task_item_or_group, dag, dag_runs, session):
+def dag_to_grid(dag, dag_runs, session):
"""
- Create a nested dict representation of this TaskGroup and its children used to construct
- the Graph.
+ Create a nested dict representation of the DAG's TaskGroup and its children
+ used to construct the Graph and Grid views.
"""
- if isinstance(task_item_or_group, AbstractOperator):
- return {
- 'id': task_item_or_group.task_id,
- 'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
- 'label': task_item_or_group.label,
- 'extra_links': task_item_or_group.extra_links,
- 'is_mapped': task_item_or_group.is_mapped,
- }
+ query = (
+ session.query(
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.state,
+ TaskInstance.start_date,
+ TaskInstance.end_date,
+ TaskInstance._try_number,
+ )
+ .filter(
+ TaskInstance.dag_id == dag.dag_id,
+ TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
+ # Only get normal task instances or the first mapped task
+ TaskInstance.map_index <= 0,
+ )
+ .order_by(TaskInstance.task_id)
+ )
- # Task Group
- task_group = task_item_or_group
+ grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
- children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
+ def task_group_to_grid(item, dag_runs, grouped_tis):
+ if isinstance(item, AbstractOperator):
- def get_summary(dag_run, children):
- child_instances = [child['instances'] for child in children if 'instances' in child]
- child_instances = [item for sublist in child_instances for item in sublist]
+ def _get_summary(task_instance):
+ try_count = (
+ task_instance._try_number
+ if task_instance._try_number != 0 or task_instance.state in State.running
+ else task_instance._try_number + 1
+ )
- children_start_dates = [item['start_date'] for item in child_instances if item]
- children_end_dates = [item['end_date'] for item in child_instances if item]
- children_states = [item['state'] for item in child_instances if item]
+ return {
+ 'task_id': task_instance.task_id,
+ 'run_id': task_instance.run_id,
+ 'map_index': task_instance.map_index,
+ 'state': task_instance.state,
+ 'start_date': task_instance.start_date,
+ 'end_date': task_instance.end_date,
+ 'try_number': try_count,
Review Comment:
Everything else seems to be fine we just need something like `mapped_states`, best format would be like so:
```
{
success: 3.
running: 2,
queued: 1,
}
```
--
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] ashb commented on pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #24284:
URL: https://github.com/apache/airflow/pull/24284#issuecomment-1148593382
Tests were done with a dag with 2000 dummy task grouped into 100 task groups
--
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] dstandish commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891422956
##########
airflow/utils/json.py:
##########
@@ -45,7 +45,7 @@ def __init__(self, *args, **kwargs):
def _default(obj):
"""Convert dates and numpy objects in a json serializable format."""
if isinstance(obj, datetime):
- return obj.strftime('%Y-%m-%dT%H:%M:%SZ')
+ return obj.isoformat()
Review Comment:
this seems to produce different output:
```
pendulum.now().timestamp()
1654540994.520137
from datetime import datetime
now = datetime.utcnow()
now.strftime('%Y-%m-%dT%H:%M:%SZ')
'2022-06-07T16:05:10Z'
now.isoformat()
'2022-06-07T16:05:10.271025'
```
do you think it matters?
--
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] bbovenzi commented on pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
bbovenzi commented on PR #24284:
URL: https://github.com/apache/airflow/pull/24284#issuecomment-1155398131
Looking good. But looks like there are some linting issues. run `yarn lint` in `airflow/www`, that should have been caught in precommit. I'll look into that
--
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] bbovenzi commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
bbovenzi commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891430623
##########
airflow/www/views.py:
##########
@@ -250,62 +251,102 @@ def _safe_parse_datetime(v):
abort(400, f"Invalid datetime: {v!r}")
-def task_group_to_grid(task_item_or_group, dag, dag_runs, session):
+def dag_to_grid(dag, dag_runs, session):
"""
- Create a nested dict representation of this TaskGroup and its children used to construct
- the Graph.
+ Create a nested dict representation of the DAG's TaskGroup and its children
+ used to construct the Graph and Grid views.
"""
- if isinstance(task_item_or_group, AbstractOperator):
- return {
- 'id': task_item_or_group.task_id,
- 'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
- 'label': task_item_or_group.label,
- 'extra_links': task_item_or_group.extra_links,
- 'is_mapped': task_item_or_group.is_mapped,
- }
+ query = (
+ session.query(
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.state,
+ TaskInstance.start_date,
+ TaskInstance.end_date,
+ TaskInstance._try_number,
+ )
+ .filter(
+ TaskInstance.dag_id == dag.dag_id,
+ TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
+ # Only get normal task instances or the first mapped task
+ TaskInstance.map_index <= 0,
+ )
+ .order_by(TaskInstance.task_id)
+ )
- # Task Group
- task_group = task_item_or_group
+ grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
- children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
+ def task_group_to_grid(item, dag_runs, grouped_tis):
+ if isinstance(item, AbstractOperator):
- def get_summary(dag_run, children):
- child_instances = [child['instances'] for child in children if 'instances' in child]
- child_instances = [item for sublist in child_instances for item in sublist]
+ def _get_summary(task_instance):
+ try_count = (
+ task_instance._try_number
+ if task_instance._try_number != 0 or task_instance.state in State.running
+ else task_instance._try_number + 1
+ )
- children_start_dates = [item['start_date'] for item in child_instances if item]
- children_end_dates = [item['end_date'] for item in child_instances if item]
- children_states = [item['state'] for item in child_instances if item]
+ return {
+ 'task_id': task_instance.task_id,
+ 'run_id': task_instance.run_id,
+ 'map_index': task_instance.map_index,
+ 'state': task_instance.state,
+ 'start_date': task_instance.start_date,
+ 'end_date': task_instance.end_date,
+ 'try_number': try_count,
Review Comment:
This no longer returns the `mapped_states` to give a status breakdown of mapped tasks. Do we still want that? If not, I need to change one bit of react 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] bbovenzi commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
bbovenzi commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891503654
##########
airflow/www/views.py:
##########
@@ -250,62 +251,102 @@ def _safe_parse_datetime(v):
abort(400, f"Invalid datetime: {v!r}")
-def task_group_to_grid(task_item_or_group, dag, dag_runs, session):
+def dag_to_grid(dag, dag_runs, session):
"""
- Create a nested dict representation of this TaskGroup and its children used to construct
- the Graph.
+ Create a nested dict representation of the DAG's TaskGroup and its children
+ used to construct the Graph and Grid views.
"""
- if isinstance(task_item_or_group, AbstractOperator):
- return {
- 'id': task_item_or_group.task_id,
- 'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
- 'label': task_item_or_group.label,
- 'extra_links': task_item_or_group.extra_links,
- 'is_mapped': task_item_or_group.is_mapped,
- }
+ query = (
+ session.query(
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.state,
+ TaskInstance.start_date,
+ TaskInstance.end_date,
+ TaskInstance._try_number,
+ )
+ .filter(
+ TaskInstance.dag_id == dag.dag_id,
+ TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
+ # Only get normal task instances or the first mapped task
+ TaskInstance.map_index <= 0,
+ )
+ .order_by(TaskInstance.task_id)
+ )
- # Task Group
- task_group = task_item_or_group
+ grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
- children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
+ def task_group_to_grid(item, dag_runs, grouped_tis):
+ if isinstance(item, AbstractOperator):
- def get_summary(dag_run, children):
- child_instances = [child['instances'] for child in children if 'instances' in child]
- child_instances = [item for sublist in child_instances for item in sublist]
+ def _get_summary(task_instance):
+ try_count = (
+ task_instance._try_number
+ if task_instance._try_number != 0 or task_instance.state in State.running
+ else task_instance._try_number + 1
+ )
- children_start_dates = [item['start_date'] for item in child_instances if item]
- children_end_dates = [item['end_date'] for item in child_instances if item]
- children_states = [item['state'] for item in child_instances if item]
+ return {
+ 'task_id': task_instance.task_id,
+ 'run_id': task_instance.run_id,
+ 'map_index': task_instance.map_index,
+ 'state': task_instance.state,
+ 'start_date': task_instance.start_date,
+ 'end_date': task_instance.end_date,
+ 'try_number': try_count,
Review Comment:
Everything else seems to be fine we just need something like `mapped_states`, best format would be like so:
```
[
{ success: 3 }.
{ running: 2 },
{ queued: 1 },
]
```
--
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] bbovenzi commented on pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
bbovenzi commented on PR #24284:
URL: https://github.com/apache/airflow/pull/24284#issuecomment-1154204558
> Stack trace for that?
>
> I didn't test with running DAGs. Oops! (Good thing we can add tests now)
```
TypeError
TypeError: '<' not supported between instances of 'NoneType' and 'NoneType'
Traceback (most recent call last)
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 2464, in __call__
def __call__(self, environ, start_response):
"""The WSGI server calls the Flask application object as the
WSGI application. This calls :meth:`wsgi_app` which can be
wrapped to applying middleware."""
return self.wsgi_app(environ, start_response)
def __repr__(self):
return "<%s %r>" % (self.__class__.__name__, self.name)
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 2450, in wsgi_app
try:
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1867, in handle_exception
# if we want to repropagate the exception, we can attempt to
# raise it with the whole traceback in case we can do that
# (the function was actually called from the except part)
# otherwise, we just raise the error again
if exc_value is e:
reraise(exc_type, exc_value, tb)
else:
raise e
self.log_exception((exc_type, exc_value, tb))
server_error = InternalServerError()
File "/usr/local/lib/python3.7/site-packages/flask/_compat.py", line 39, in reraise
import collections.abc as collections_abc
def reraise(tp, value, tb=None):
if value.__traceback__ is not tb:
raise value.with_traceback(tb)
raise value
implements_to_string = _identity
else:
iterkeys = lambda d: d.iterkeys()
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 2447, in wsgi_app
ctx = self.request_context(environ)
error = None
try:
try:
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1952, in full_dispatch_request
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
def finalize_request(self, rv, from_error_handler=False):
"""Given the return value from a view function this finalizes
the request by converting it into a response and invoking the
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1821, in handle_user_exception
return self.handle_http_exception(e)
handler = self._find_error_handler(e)
if handler is None:
reraise(exc_type, exc_value, tb)
return handler(e)
def handle_exception(self, e):
"""Handle an exception that did not have an error handler
associated with it, or that was raised from an error handler.
File "/usr/local/lib/python3.7/site-packages/flask/_compat.py", line 39, in reraise
import collections.abc as collections_abc
def reraise(tp, value, tb=None):
if value.__traceback__ is not tb:
raise value.with_traceback(tb)
raise value
implements_to_string = _identity
else:
iterkeys = lambda d: d.iterkeys()
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1950, in full_dispatch_request
self.try_trigger_before_first_request_functions()
try:
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
def finalize_request(self, rv, from_error_handler=False):
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1936, in dispatch_request
getattr(rule, "provide_automatic_options", False)
and req.method == "OPTIONS"
):
return self.make_default_options_response()
# otherwise dispatch to the handler for that endpoint
return self.view_functions[rule.endpoint](**req.view_args)
def full_dispatch_request(self):
"""Dispatches the request and on top of that performs request
pre and postprocessing as well as HTTP exception catching and
error handling.
File "/opt/airflow/airflow/www/auth.py", line 43, in decorated
dag_id = (
request.args.get("dag_id") or request.form.get("dag_id") or (request.json or {}).get("dag_id")
)
if appbuilder.sm.check_authorization(permissions, dag_id):
return func(*args, **kwargs)
elif not g.user.is_anonymous and not g.user.perms:
return (
render_template(
'airflow/no_roles_permissions.html',
hostname=socket.getfqdn()
File "/opt/airflow/airflow/www/views.py", line 3622, in grid_data
dag_runs = query.order_by(DagRun.execution_date.desc()).limit(num_runs).all()
dag_runs.reverse()
encoded_runs = [wwwutils.encode_dag_run(dr) for dr in dag_runs]
data = {
'groups': dag_to_grid(dag, dag_runs, session),
'dag_runs': encoded_runs,
}
# avoid spaces to reduce payload size
return (
File "/opt/airflow/airflow/www/views.py", line 396, in dag_to_grid
'children': children,
'tooltip': task_group.tooltip,
'instances': group_summaries,
}
return task_group_to_grid(dag.task_group, dag_runs, grouped_tis)
def task_group_to_dict(task_item_or_group):
"""
Create a nested dict representation of this TaskGroup and its children used to construct
File "/opt/airflow/airflow/www/views.py", line 357, in task_group_to_grid
# Task Group
task_group = item
children = [
task_group_to_grid(child, dag_runs, grouped_tis) for child in task_group.topological_sort()
]
def get_summary(dag_run, children):
child_instances = [child['instances'] for child in children if 'instances' in child]
child_instances = [
File "/opt/airflow/airflow/www/views.py", line 357, in <listcomp>
# Task Group
task_group = item
children = [
task_group_to_grid(child, dag_runs, grouped_tis) for child in task_group.topological_sort()
]
def get_summary(dag_run, children):
child_instances = [child['instances'] for child in children if 'instances' in child]
child_instances = [
File "/opt/airflow/airflow/www/views.py", line 341, in task_group_to_grid
if record:
set_overall_state(record)
yield record
if item.is_mapped:
instances = list(_mapped_summary(grouped_tis.get(item.task_id, [])))
else:
instances = list(map(_get_summary, grouped_tis.get(item.task_id, [])))
return {
'id': item.task_id,
File "/opt/airflow/airflow/www/views.py", line 333, in _mapped_summary
'end_date': ti_summary.end_date,
'mapped_states': {ti_summary.state: ti_summary.state_count},
'state': None, # We change this before yielding
}
continue
record['start_date'] = min(record['start_date'], ti_summary.start_date)
record['end_date'] = max(record['end_date'], ti_summary.end_date)
record['mapped_states'][ti_summary.state] = ti_summary.state_count
if record:
set_overall_state(record)
yield record
TypeError: '<' not supported between instances of 'NoneType' and 'NoneType'
--
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] ashb commented on pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #24284:
URL: https://github.com/apache/airflow/pull/24284#issuecomment-1155286182
PTAL @bbovenzi I've fixed the issue
--
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] ashb commented on pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #24284:
URL: https://github.com/apache/airflow/pull/24284#issuecomment-1148591575
(I'm still not thrilled at it taking 1.5s mind you, but it's a darn sight better)
--
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] ashb commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891501375
##########
airflow/www/views.py:
##########
@@ -250,62 +251,102 @@ def _safe_parse_datetime(v):
abort(400, f"Invalid datetime: {v!r}")
-def task_group_to_grid(task_item_or_group, dag, dag_runs, session):
+def dag_to_grid(dag, dag_runs, session):
"""
- Create a nested dict representation of this TaskGroup and its children used to construct
- the Graph.
+ Create a nested dict representation of the DAG's TaskGroup and its children
+ used to construct the Graph and Grid views.
"""
- if isinstance(task_item_or_group, AbstractOperator):
- return {
- 'id': task_item_or_group.task_id,
- 'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
- 'label': task_item_or_group.label,
- 'extra_links': task_item_or_group.extra_links,
- 'is_mapped': task_item_or_group.is_mapped,
- }
+ query = (
+ session.query(
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.state,
+ TaskInstance.start_date,
+ TaskInstance.end_date,
+ TaskInstance._try_number,
+ )
+ .filter(
+ TaskInstance.dag_id == dag.dag_id,
+ TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
+ # Only get normal task instances or the first mapped task
+ TaskInstance.map_index <= 0,
+ )
+ .order_by(TaskInstance.task_id)
+ )
- # Task Group
- task_group = task_item_or_group
+ grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
- children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
+ def task_group_to_grid(item, dag_runs, grouped_tis):
+ if isinstance(item, AbstractOperator):
- def get_summary(dag_run, children):
- child_instances = [child['instances'] for child in children if 'instances' in child]
- child_instances = [item for sublist in child_instances for item in sublist]
+ def _get_summary(task_instance):
+ try_count = (
+ task_instance._try_number
+ if task_instance._try_number != 0 or task_instance.state in State.running
+ else task_instance._try_number + 1
+ )
- children_start_dates = [item['start_date'] for item in child_instances if item]
- children_end_dates = [item['end_date'] for item in child_instances if item]
- children_states = [item['state'] for item in child_instances if item]
+ return {
+ 'task_id': task_instance.task_id,
+ 'run_id': task_instance.run_id,
+ 'map_index': task_instance.map_index,
+ 'state': task_instance.state,
+ 'start_date': task_instance.start_date,
+ 'end_date': task_instance.end_date,
+ 'try_number': try_count,
Review Comment:
Oh
```diff
- if task_instance.map_index > -1:
- return get_mapped_summary(
- task_instance, task_instances=get_mapped_instances(task_instance, session)
- )
```
--
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] bbovenzi commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
bbovenzi commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891576098
##########
airflow/www/views.py:
##########
@@ -250,62 +251,102 @@ def _safe_parse_datetime(v):
abort(400, f"Invalid datetime: {v!r}")
-def task_group_to_grid(task_item_or_group, dag, dag_runs, session):
+def dag_to_grid(dag, dag_runs, session):
"""
- Create a nested dict representation of this TaskGroup and its children used to construct
- the Graph.
+ Create a nested dict representation of the DAG's TaskGroup and its children
+ used to construct the Graph and Grid views.
"""
- if isinstance(task_item_or_group, AbstractOperator):
- return {
- 'id': task_item_or_group.task_id,
- 'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
- 'label': task_item_or_group.label,
- 'extra_links': task_item_or_group.extra_links,
- 'is_mapped': task_item_or_group.is_mapped,
- }
+ query = (
+ session.query(
+ TaskInstance.task_id,
+ TaskInstance.run_id,
+ TaskInstance.map_index,
+ TaskInstance.state,
+ TaskInstance.start_date,
+ TaskInstance.end_date,
+ TaskInstance._try_number,
+ )
+ .filter(
+ TaskInstance.dag_id == dag.dag_id,
+ TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
+ # Only get normal task instances or the first mapped task
+ TaskInstance.map_index <= 0,
+ )
+ .order_by(TaskInstance.task_id)
+ )
- # Task Group
- task_group = task_item_or_group
+ grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
- children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
+ def task_group_to_grid(item, dag_runs, grouped_tis):
+ if isinstance(item, AbstractOperator):
- def get_summary(dag_run, children):
- child_instances = [child['instances'] for child in children if 'instances' in child]
- child_instances = [item for sublist in child_instances for item in sublist]
+ def _get_summary(task_instance):
+ try_count = (
+ task_instance._try_number
+ if task_instance._try_number != 0 or task_instance.state in State.running
+ else task_instance._try_number + 1
+ )
- children_start_dates = [item['start_date'] for item in child_instances if item]
- children_end_dates = [item['end_date'] for item in child_instances if item]
- children_states = [item['state'] for item in child_instances if item]
+ return {
+ 'task_id': task_instance.task_id,
+ 'run_id': task_instance.run_id,
+ 'map_index': task_instance.map_index,
+ 'state': task_instance.state,
+ 'start_date': task_instance.start_date,
+ 'end_date': task_instance.end_date,
+ 'try_number': try_count,
Review Comment:
We can make the decision on the server. We were giving an overall status and a breakdown of the status of all the children that form that overall status. But we can just do an overall status if you'd like
--
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] dstandish commented on a diff in pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #24284:
URL: https://github.com/apache/airflow/pull/24284#discussion_r891629900
##########
airflow/utils/json.py:
##########
@@ -45,7 +45,7 @@ def __init__(self, *args, **kwargs):
def _default(obj):
"""Convert dates and numpy objects in a json serializable format."""
if isinstance(obj, datetime):
- return obj.strftime('%Y-%m-%dT%H:%M:%SZ')
+ return obj.isoformat()
Review Comment:
But the original code also stripped out sub-second precision and isoformat includes micros, it appears. Don't know whether that matters here.
--
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] bbovenzi commented on pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
bbovenzi commented on PR #24284:
URL: https://github.com/apache/airflow/pull/24284#issuecomment-1154192240
I'm getting this error during auto-refresh:
`TypeError: '<' not supported between instances of 'datetime.datetime' and 'NoneType'`
--
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] ashb commented on pull request #24284: Speed up grid_data endpoint by 10x
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #24284:
URL: https://github.com/apache/airflow/pull/24284#issuecomment-1156384088
Helm/Kube tests are not caused by this PR and are failing on main too. Merging.
--
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