Add support for labelling DAG edges #15142

andrewgodwin · 2021-04-01T17:56:59Z

This adds support for putting human-readable labels on edges in the DAG between Tasks, as well as making the underlying framework for that generic enough that future metadata could be added if desired.

What's left to do:

It modifies both the GraphViz and the D3 renderers - example:

from airflow.models import Label
...
data = fetch_data()
mean_height = calculate_mean_height(data)
mean_height >> Label("Mean") >> print_result
max_height = calculate_max_height(data)
max_height.set_downstream(print_result_again, Label("Max"))

closes: #15140

ashb · 2021-04-01T18:03:37Z

from airflow.models import Label

I'm not a fan of this import for in user dags (they generally don't import from models currently) but I don't have an immediate other suggestion.

(Just a comment on "user" import not code location)

andrewgodwin · 2021-04-01T18:05:29Z

from airflow.models import Label

I'm not a fan of this import for in user dags (they generally don't import from models currently) but I don't have an immediate other suggestion

Yeah, I went trawling around some example DAGs to find what path might fit, and I saw a couple with models imports for Variable and so forth, so I went with that initially. I'm not sure what airflow.task is for - seems pretty empty - but that was something else that felt feasible.

ashb · 2021-04-01T18:13:18Z

To add to your Todo list: update the SerialisedDag representations to include this

andrewgodwin · 2021-04-01T18:19:48Z

To add to your Todo list: update the SerialisedDag representations to include this

That's actually done already, or it wouldn't make it across to the UI. Unless there's extra work above and beyond just making it appear in the schema & json?

ashb · 2021-04-01T18:21:34Z

To add to your Todo list: update the SerialisedDag representations to include this

That's actually done already, or it wouldn't make it across to the UI. Unless there's extra work above and beyond just making it appear in the schema & json?

Nope, I just missed it then! GitHub mobile is not the best at displaying PRs

ashb · 2021-04-01T18:23:16Z

airflow/serialization/schema.json

@@ -102,7 +102,8 @@
        "_task_group": {"anyOf": [
          { "type": "null" },
          { "$ref": "#/definitions/task_group" }
-        ]}
+        ]},
+        "edge_info": { "$ref": "#/definitions/dict" }


Oh this could maybe be specified a bit tighter. WDYT @kaxil ?

Yea similar to how we only store _downstream_task_ids (not the _upstream_task_ids) for a task and we can set upstream for the other when de-serializing. We do that to keep the size of serialized blog as small as possible.

airflow/airflow/serialization/serialized_objects.py

Lines 451 to 452 in 8cc8d11

if k == "_downstream_task_ids":

v = set(v)

airflow/airflow/serialization/serialized_objects.py

Lines 666 to 668 in 8cc8d11

for k, v in encoded_dag.items():

if k == "_downstream_task_ids":

v = set(v)

airflow/airflow/serialization/serialized_objects.py

Lines 724 to 727 in 8cc8d11

for task_id in serializable_task.downstream_task_ids:

# Bypass set_upstream etc here - it does more than we want

# noqa: E501 # pylint: disable=protected-access

dag.task_dict[task_id]._upstream_task_ids.add(serializable_task.task_id)

airflow/airflow/models/baseoperator.py

Lines 1365 to 1370 in 8cc8d11

cls.__serialized_fields = frozenset(

vars(BaseOperator(task_id='test')).keys()

- {

'inlets',

'outlets',

'_upstream_task_ids',

We pay the price of hardcoding it but it is worth it as we can save MBs (which would be transmitted to and from the database) when number of tasks is huge.

Sure, I'll add a slightly more compact representation. I can't save any actual data, since source-target-label is all that's stored, but we can save the outside wrapping with _type etc. at least?

uranusjr · 2021-04-01T19:10:18Z

from airflow.models import Label

How about hiding the Label class from the user entirely? Something like task1 << "Label" << task2 and create the label automatically in _set_relatives().

andrewgodwin · 2021-04-01T20:11:00Z

from airflow.models import Label

How about hiding the Label class from the user entirely? Something like task1 << "Label" << task2 and create the label automatically in _set_relatives().

I'd not be a huge fan of that design for two reasons:

If we ever want to add edge information beyond just a label (e.g. add a longer description), it means we have to go back to this pattern
It's relatively unobvious to new people reading the code what the meaning of a plain string is, whereas I think wrapping it in Label gives it immediate meaning to the casual reader

xinbinhuang · 2021-04-02T05:26:19Z

If we ever want to add edge information beyond just a label (e.g. add a longer description), it means we have to go back to this pattern

Can you give an example of a long description that has to use Label but not otherwise?

How about task1 >> "this is a label" | task2? This is the exact opposite of the Beam Python SDK example here.

Though I am a bit hesitant to introduce another special operator for Airflow, I think it's still better rather than introducing a Label inheriting from TaskMixin for these 2 reasons:

Conceptually, label and task should not be treated the same. The label is metadata but the task is a task. They are different in both UI and execution.
In general, I don't like the explicit import of UI components in DAG code. I think DAG code should focus on execution/dependency logic

andrewgodwin · 2021-04-02T14:03:00Z

I would argue that this fulfills a similar role as task groups - it's a primarily UI-focused feature, but it's also DAG metadata and should live in it. If it's not in the DAG, where would it go where it can appear both in the webserver view and in the output from airflow dags show?

I'm open to changing the user interface here - this is just a rough idea me and @ashb discussed - but I really don't like moving to just a plain str as the thing we use with >> as it's going to make the code significantly more complex to track where everything is supposed to go, especially when combined with XComArgs (which are the model I based this on). Not to mention all the typing for >> expects something inheriting from TaskMixin right now.

I'm also not sure you could make this work:

[t1, t2, t3] >> "label" >> [t4, t5, t6]

uranusjr · 2021-04-02T16:49:56Z

I'm also not sure you could make this work:
[t1, t2, t3] >> "label" >> [t4, t5, t6]

I don’t think [t1, t2, t3] >> [t4, t5, t6] works either, so that’s not a big problem IMO. But I am currently in camp explicit Label now although I proposed the str syntax; there are things a plain str can’t handle, so a class will be needed sooner or later.

turbaszek · 2021-04-02T17:03:23Z

When doing operators for adding labels it would be good to check if they work with lineage operators (> and < afaik).

@andrewgodwin would you mind elaborating more about the use case for the labels? In my understanding all the actions are performed by tasks and edges in DAGs represent only order (and sometimes data) relation. Adding a label to edge would suggest that some action is performed between tasks. Is it done by XCom or something?

andrewgodwin · 2021-04-02T17:24:34Z

The idea for labels is merely to give the user a visual indication of what each edge means - this can be especially important, IMO, when using the branching operators or similar, on larger dags. It has no runtime effect.

I don't think I'd even want to add anything with a runtime effect on edges; Airflow is designed around all the runtime info (priority, etc.) being at the Task level and I like that. The only things I could forsee adding is other informational data, such as a longer "description" field.

The other option to achieve this is to take this, task groups, and any other informational-only parts of airflow and spin them off into a separate presentation layer somehow, and while I do quite like separating presentation and logic, I think that would be too unwieldy to actually be useful.

airflow/models/edgemodifier.py

airflow/utils/types.py

kaxil · 2021-04-02T22:57:46Z

airflow/serialization/schema.json

@@ -102,7 +102,8 @@
        "_task_group": {"anyOf": [
          { "type": "null" },
          { "$ref": "#/definitions/task_group" }
-        ]}
+        ]},
+        "edge_info": { "$ref": "#/definitions/dict" }


Yea similar to how we only store _downstream_task_ids (not the _upstream_task_ids) for a task and we can set upstream for the other when de-serializing. We do that to keep the size of serialized blog as small as possible.

airflow/airflow/serialization/serialized_objects.py

Lines 451 to 452 in 8cc8d11

if k == "_downstream_task_ids":

v = set(v)

airflow/airflow/serialization/serialized_objects.py

Lines 666 to 668 in 8cc8d11

for k, v in encoded_dag.items():

if k == "_downstream_task_ids":

v = set(v)

airflow/airflow/serialization/serialized_objects.py

Lines 724 to 727 in 8cc8d11

for task_id in serializable_task.downstream_task_ids:

# Bypass set_upstream etc here - it does more than we want

# noqa: E501 # pylint: disable=protected-access

dag.task_dict[task_id]._upstream_task_ids.add(serializable_task.task_id)

airflow/airflow/models/baseoperator.py

Lines 1365 to 1370 in 8cc8d11

cls.__serialized_fields = frozenset(

vars(BaseOperator(task_id='test')).keys()

- {

'inlets',

'outlets',

'_upstream_task_ids',

We pay the price of hardcoding it but it is worth it as we can save MBs (which would be transmitted to and from the database) when number of tasks is huge.

andrewgodwin · 2021-04-08T17:18:12Z

(ignore it being ready for review, I clicked that before remembering the serialisation needs fixing)

airflow/www/templates/airflow/graph.html

github-actions · 2021-04-08T17:34:40Z

The Workflow run is cancelling this PR. Building images for the PR has failed. Follow the workflow link to check the reason.

ryanahamilton · 2021-04-08T20:54:50Z

There are a couple of visual tweaks we should add when hovering tasks or statuses (e.g. fading out labels ). I'll provide you a solution as I was just modifying these interactions in #15257.

kaxil · 2021-04-08T23:45:44Z

docs/apache-airflow/concepts.rst

+As well as grouping tasks into groups, you can also label the edges between
+different tasks in the Graph View - this can be especially useful for branching
+areas of your DAG, so you can label the conditions under which certain branches
+might run.


nit: Probably a screenshot in this doc might be helpful

Don't know how I didn't think about that one given it's an entirely visual feature! I'll get it in tomorrow.

I'm gonna merge this PR so we don't end up in CI hell, but @andrewgodwin please make another PR with this screenshot if you think that will help

@andrewgodwin can you also add some Edge Labels to a few of the DAGs in airflow/example_dags/ in the same PR? It would be useful for this feature to be exposed during local development.

github-actions · 2021-04-09T00:11:40Z

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 master at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

This adds support for putting human-readable labels on edges in the DAG between Tasks, as well as making the underlying framework for that generic enough that future metadata could be added if desired.

ryanahamilton · 2021-04-09T02:24:20Z

I opened #15298 to address the aforementioned visual enhancements needed for this feature.

Part of the resolution of #15140 (paired with #15142) "Edge labels" are an existing concept in the library that powers the Graph view. This feature will be employed in Airflow the Airflow Graph view with #15142. This PR ensures that the labels are displayed properly when interacting with the Task/path/status highlighting features of the Graph. Primarily, it fades out the labels when not relevant.

andrewgodwin · 2021-04-09T17:27:24Z

I've opened #15310 to get the extra examples in.

boring-cyborg bot added area:serialization area:UI Related to UI/UX. For Frontend Developers. area:webserver Webserver related Issues labels Apr 1, 2021

ashb reviewed Apr 1, 2021

View reviewed changes

andrewgodwin force-pushed the edge-labels branch from 6201e42 to dffd368 Compare April 2, 2021 18:09

kaxil requested changes Apr 2, 2021

View reviewed changes

andrewgodwin force-pushed the edge-labels branch from dffd368 to 7956f6c Compare April 5, 2021 15:13

dimberman self-requested a review April 5, 2021 17:59

andrewgodwin force-pushed the edge-labels branch from 7956f6c to 743a31a Compare April 8, 2021 17:15

andrewgodwin marked this pull request as ready for review April 8, 2021 17:16

andrewgodwin requested review from mik-laj, ryanahamilton, turbaszek, vikramkoka and XD-DENG as code owners April 8, 2021 17:16

andrewgodwin changed the title ~~Add support for labelling DAG edges (WIP)~~ Add support for labelling DAG edges Apr 8, 2021

ryanahamilton reviewed Apr 8, 2021

View reviewed changes

airflow/www/templates/airflow/graph.html Outdated Show resolved Hide resolved

andrewgodwin force-pushed the edge-labels branch 2 times, most recently from bbbdaff to b80d879 Compare April 8, 2021 19:33

kaxil reviewed Apr 8, 2021

View reviewed changes

kaxil approved these changes Apr 9, 2021

View reviewed changes

github-actions bot added the full tests needed We need to run full set of tests for this PR to merge label Apr 9, 2021

Add support for labelling DAG edges

18e3b8b

This adds support for putting human-readable labels on edges in the DAG between Tasks, as well as making the underlying framework for that generic enough that future metadata could be added if desired.

dimberman force-pushed the edge-labels branch from b80d879 to 18e3b8b Compare April 9, 2021 01:17

ryanahamilton mentioned this pull request Apr 9, 2021

Graph "edge label" interaction styling #15298

Merged

dimberman merged commit 19b74fd into apache:master Apr 9, 2021

andrewgodwin deleted the edge-labels branch April 9, 2021 17:18

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add support for labelling DAG edges #15142

Add support for labelling DAG edges #15142

andrewgodwin commented Apr 1, 2021 •

edited

Loading

ashb commented Apr 1, 2021 •

edited

Loading

andrewgodwin commented Apr 1, 2021

ashb commented Apr 1, 2021

andrewgodwin commented Apr 1, 2021

ashb commented Apr 1, 2021

ashb Apr 1, 2021

kaxil Apr 2, 2021

andrewgodwin Apr 5, 2021

uranusjr commented Apr 1, 2021

andrewgodwin commented Apr 1, 2021

xinbinhuang commented Apr 2, 2021 •

edited

Loading

andrewgodwin commented Apr 2, 2021 •

edited

Loading

uranusjr commented Apr 2, 2021

turbaszek commented Apr 2, 2021 •

edited

Loading

andrewgodwin commented Apr 2, 2021 •

edited

Loading

kaxil Apr 2, 2021

andrewgodwin commented Apr 8, 2021

github-actions bot commented Apr 8, 2021

ryanahamilton commented Apr 8, 2021

kaxil Apr 8, 2021

andrewgodwin Apr 9, 2021

dimberman Apr 9, 2021

ryanahamilton Apr 9, 2021

github-actions bot commented Apr 9, 2021

ryanahamilton commented Apr 9, 2021

andrewgodwin commented Apr 9, 2021

	for k, v in encoded_dag.items():
	if k == "_downstream_task_ids":
	v = set(v)

	for task_id in serializable_task.downstream_task_ids:
	# Bypass set_upstream etc here - it does more than we want
	# noqa: E501 # pylint: disable=protected-access
	dag.task_dict[task_id]._upstream_task_ids.add(serializable_task.task_id)

	cls.__serialized_fields = frozenset(
	vars(BaseOperator(task_id='test')).keys()
	- {
	'inlets',
	'outlets',
	'_upstream_task_ids',

Add support for labelling DAG edges #15142

Add support for labelling DAG edges #15142

Conversation

andrewgodwin commented Apr 1, 2021 • edited Loading

ashb commented Apr 1, 2021 • edited Loading

andrewgodwin commented Apr 1, 2021

ashb commented Apr 1, 2021

andrewgodwin commented Apr 1, 2021

ashb commented Apr 1, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

uranusjr commented Apr 1, 2021

andrewgodwin commented Apr 1, 2021

xinbinhuang commented Apr 2, 2021 • edited Loading

andrewgodwin commented Apr 2, 2021 • edited Loading

uranusjr commented Apr 2, 2021

turbaszek commented Apr 2, 2021 • edited Loading

andrewgodwin commented Apr 2, 2021 • edited Loading

Choose a reason for hiding this comment

andrewgodwin commented Apr 8, 2021

github-actions bot commented Apr 8, 2021

ryanahamilton commented Apr 8, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Apr 9, 2021

ryanahamilton commented Apr 9, 2021

andrewgodwin commented Apr 9, 2021

andrewgodwin commented Apr 1, 2021 •

edited

Loading

ashb commented Apr 1, 2021 •

edited

Loading

xinbinhuang commented Apr 2, 2021 •

edited

Loading

andrewgodwin commented Apr 2, 2021 •

edited

Loading

turbaszek commented Apr 2, 2021 •

edited

Loading

andrewgodwin commented Apr 2, 2021 •

edited

Loading