Order of predecessors to a node can matter

[dmentipl]

When a node has multiple predecessors the order in which they are defined and then passed to create the networkx DiGraph matters. However, it's not clear to me that it should matter for a directed acyclic graph. As far as I can tell, for a collection of predecessors for a node in a DAG has no notion of ordering of those predecessors (however, instead, they could be labelled).

For example, in the demo notebook in cell 3 if you swap the order:

    # node2 -> node5
    node5 = Node(step="step5", leadtime=leadtime)
    EDGES.append((node2, node5))

    # node4 -> node5
    node4 = Node(step="step4", leadtime=leadtime)
    EDGES.append((node4, node5))

for

    # node4 -> node5
    node4 = Node(step="step4", leadtime=leadtime)
    EDGES.append((node4, node5))

    # node2 -> node5
    node5 = Node(step="step5", leadtime=leadtime)
    EDGES.append((node2, node5))

Then the result for node5 (see cell 8) is 3_4_1_2_5 not 1_2_3_4_5.

This is an issue that I believe is inherited from paraflow, in a sense. I came across this in the context of paraflow while experimenting with filtering the DiGraph object (for hindcasting) and recreating a new DiGraph object messed up the order. For the IMPROVER suite the ${INPUT} argument matters.

I think this is something to consider in the design. I understand that in the case of the demo notebook if the concatenate_id function were defined differently this problem could be removed. However it still could lead to some confusion. And, in general, we cannot rely on the ordering of the predecessors to a node, especially if we generate and manipulate the DiGraph object itself.

Carwyn Pelley (@cpelley) (and others): I'd be interested to hear your thoughts on this.

---

As an aside, I'm investigating the use of DAGRunner for a verification workflow at the Bureau of Meteorology. The use of DAGRunner would be advantageous as it allows for multiple schedulers rather than just Cylc as for paraflow/

[cpelley]

Thanks for your interest Daniel Mentiplay (@dmentipl).

...it's not clear to me that it should matter for a directed acyclic graph
I think this is something to consider in the design.

Yes and no. Let's say two nodes connect to a third.

graph

A & B --> C

Loading

A DAG doesn't inherently characterise the order of connectivity, only that there are these connections.
When it comes to it's execution, it's then the role and responsibility of the application called in C to be made input order agnostic.

DAGrunner has no opinion to how a project/user defines its graphs. It's role and responsibility is in understanding how to execute networkx graphs.
The example application (concatenate_id defined in this case) in the notebook is specifically intended to be order agnostic, showing the exact flow of data between nodes in the execution, thereby relaying some intuition to users unfamiliar to graph execution.

With regard to post-processing graphs, defining and using applications with order depended arguments would indeed generally be bad idea - and we are slowly making moves to rectify this in improver applications.
It isn't the role of DAGrunner to define standards for defining networkx graphs for processing. However, https://github.com/MetOffice/pp_demo_workflow is. You might want to look there about defining configurations.

[dmentipl]

Thanks for your response Carwyn Pelley (@cpelley).

A DAG doesn't inherently characterise the order of connectivity, only that there are these connections. When it comes to it's execution, it's then the role and responsibility of the application called in C to be made input order agnostic. DAGrunner has no opinion to how a project/user defines its graphs. It's role and responsibility is in understanding how to execute networkx graphs.

Yep, this all makes sense, and agrees with my intuition here.

The example application (concatenate_id defined in this case) in the notebook is specifically intended to be order agnostic, showing the exact flow of data between nodes in the execution, thereby relaying some intuition to users unfamiliar to graph execution.

I think in this case it's not "order agnostic" but "order dependent" (which is on purpose for the demo notebook as you say).

With regard to post-processing graphs, defining and using applications with order depended arguments would indeed generally be bad idea - and we are slowly making moves to rectify this in improver applications. It isn't the role of DAGrunner to define standards for defining networkx graphs for processing. However, https://github.com/MetOffice/pp_demo_workflow is. You might want to look there about defining configurations.

I would like to take a look at pp_demo_workflow. I don't currently have access. Could you provide access?

[cpelley]

I think in this case it's not "order agnostic" but "order dependent" (which is on purpose for the demo notebook as you say).

Sorry, yes I meant order dependent here :)

I would like to take a look at pp_demo_workflow. I don't currently have access. Could you provide access?

Sure

[dmentipl]

Hi Carwyn Pelley (@cpelley), thank for your previous responses and for access to pp_demo_workflow.

I'm still trying to wrap my head around this. Digging into execute_graph.py we have

        exec_graph = {}
        for node_id, properties in self._nxgraph.nodes(data=True):
            # don't use nodes in our graph as some schedulers (dask
            # distributed as per dask.core.validate_key) support only a subset
            # of types (tuples, bytes, int, float and str).
            key = tokenize(node_id)
            args = [tokenize(arg) for arg in self._nxgraph.predecessors(node_id)]
            exec_graph[key] = (apply, executor, args, properties | {"node_id": node_id})

from dagrunner/execute_graph.py#L368-L375. When setting the args variable we are relying on the networkx DiGraph.predecessors method which could return the predecessors in any order as a DAG has no notion of ordering of predecessors.

Say we are using the improver cli (or plugin) to apply a lapse rate correction. The cli/function signature is:

def process(
    temperature: cli.inputcube,
    lapse_rate: cli.inputcube,
    source_orography: cli.inputcube,
    target_orography: cli.inputcube,
):

from improver/cli/apply_lapse_rate.py#L13-L18. We need to supply the args in the correct order, or use keyword arguments.

In our IMPROVER workflow we have a graph like:

# Standard processing
EDGES.extend(
    linear_nodechain(
        [
            Node(system="impr", model="bom_access_g4", chain="tempscreen", step="standardise", leadtime=leadtime, cycle=cycle),
            Node(system="impr", model="bom_access_g4", chain="tempscreen", step="regrid", leadtime=leadtime, cycle=cycle),
            Node(system="impr", model="bom_access_g4", chain="tempscreen", step="apply_gridded_lapse_rate", leadtime=leadtime, cycle=cycle),
            Node(system="impr", model="bom_access_g4", chain="tempscreen", step="apply_bias_correction", leadtime=leadtime, cycle=cycle),
            Node(system="impr", model="bom_access_g4", chain="tempscreen", step="threshold", leadtime=leadtime, cycle=cycle),
            Node(system="impr", model="bom_access_g4", chain="tempscreen", step="nb_topographic", leadtime=leadtime, cycle=cycle),
            Node(system="impr", model="bom_access_g4", chain="tempscreen", step="recfilter", leadtime=leadtime, cycle=cycle),
            Node(system="impr", model="bom_access_g4", chain="tempscreen", step="apply_reliability_calibration", leadtime=leadtime, cycle=cycle),
            Node(system="impr", model="bom_access_g4", chain="tempscreen", step="level2", leadtime=leadtime, cycle=cycle),
        ]
    )
)

# Lapse rate
EDGES.append(
    [
        Node(system="impr", model="bom_access_g4", chain="lapserate", step="calc_lapse_rate", leadtime=leadtime, cycle=cycle),
        Node(system="impr", model="bom_access_g4", chain="tempscreen", step="apply_gridded_lapse_rate", leadtime=leadtime, cycle=cycle),
    ]
)

EDGES.extend(
    many_to_one_nodechain(
        [
            Node(system="ancil", model="bom_access_g4", chain="orography", step=f"model_on_{GRID}"),
            Node(system="ancil", chain="orography", step=GRID),
        ],
        Node(system="impr", model="bom_access_g4", chain="tempscreen", step="apply_gridded_lapse_rate", leadtime=leadtime, cycle=cycle),
    )
)

Thus we have defined the dependencies in the order required by the improver cli/function/plugin, as required. Note: we are using paraflow, of course, but I expect that we can use a similar graph definition using DAGRunner.

However, I'm still unsure how the order issue is resolved. With dask.Task it seems like you specifically define the task with its args supplied as references to other tasks, as in:

t = Task("key", add, TaskRef("a"), TaskRef("b"))

See Dask docs. (I suppose "add" is probably not a good example as order doesn't matter but you get the idea.)

Given that the args in DAGRunner are supplied as above via a call to DiGraph.predecessors I can't see how any task using args derived from graph dependencies can reliably determine which argument is which.

As a very simple example, imagine a function/plugin/task that chooses the first of two inputs:

from dataclass import dataclass

@dataclass
class Node:
    step: str

def first(a, b):
    return a

EDGES = [
    (Node(step="left"), Node(step="target")),
    (Node(step="right"), Node(step="target")),
]

SETTINGS = {
    Node(step="left"): {"call": (id, {})},
    Node(step="right"): {"call": (id, {})},
    Node(step="target"): {"call": (first, {})},
}

Should it matter if we write the following instead?

EDGES = [
    (Node(step="right"), Node(step="target")),
    (Node(step="left"), Node(step="target")),
]

I'd be interested to know what you think. Maybe I'm completely missing something.

[cpelley]

Apologies Daniel Mentiplay (@dmentipl). I think we are perhaps talking passed one another. I fully agree with the issue with existing paraflow workflows, it's just that we see the solution to complexity, transparency and maintainability as moving responsibilities to the application level, rather than solving them at the graph level (i.e. by application I mean plugins - trying to move away from the term).
Whilst most problems can be treated as graph problems, things become difficult to maintain, understand and support as it scales and changes over time.

your example, using metadata to identify which is which:

def process(self, *cubes: Union[Cube, CubeList]) -> Cube:
    ...
    cubes = as_cubelist(*cubes)
    (
        temp_cube
        lapse_rate
        ...
    ) = cubes.extract(
        [
            "temperature...",
            "lapse_rate...",
            ....
        ]
    ...

There is of course some nuance here as there are cases where metadata may not currently we available to differentiate one from another or other cases where additional arguments are needed to control what metadata is used to pick one from another etc.

---

Further examples to moving responsibility from the graph to the application:
https://github.com/MetOffice/pp_demo_workflow/pull/70

Here, we envisage a blending workflow where we move the responsibility of what should be blended and whether to blend, to applications for that purpose. That is, no longer handling that complexity within the graph (which is incredibly complex in our case at least).

[cpelley]

To further clarify, there is nothing stopping projects from defining NetworkX graphs that are reliant on insertion order or otherwise defining some approach that determines this insertion order when constructing their NetworkX graphs. It's just that unlike Paraflow, DAGrunner doesn't concern itself with graph construction (it is simply passed a NetworkX graph or your callable that returns one).

Looking back to the notebook example:

    EDGES.append((node2, node5))
    EDGES.append((node4, node5))

graph TD

0(["0<br>S:step1_L:0"])
click 0 callback "step: 'step1'<br>leadtime: 0<br>call: (#lt;function concatenate_id at 0x7f3ecc718ea0#gt;, {'id': 1})"
1(["1<br>S:step2_L:0"])
click 1 callback "step: 'step2'<br>leadtime: 0<br>call: (#lt;function concatenate_id at 0x7f3ecc718ea0#gt;, {'id': 2})"
2(["2<br>S:step3_L:0"])
click 2 callback "step: 'step3'<br>leadtime: 0<br>call: (#lt;function concatenate_id at 0x7f3ecc718ea0#gt;, {'id': 3})"
3(["3<br>S:step4_L:0"])
click 3 callback "step: 'step4'<br>leadtime: 0<br>call: (#lt;function concatenate_id at 0x7f3ecc718ea0#gt;, {'id': 4})"
4(["4<br>S:step5_L:0"])
click 4 callback "step: 'step5'<br>leadtime: 0<br>call: (#lt;function concatenate_id at 0x7f3ecc718ea0#gt;, {'id': 5})"
0 --> 1
2 --> 3
1 --> 4
3 --> 4

Loading

    EDGES.append((node4, node5))
    EDGES.append((node2, node5))

graph TD

0(["0<br>S:step1_L:0"])
click 0 callback "step: 'step1'<br>leadtime: 0<br>call: (#lt;function concatenate_id at 0x7faa06e21800#gt;, {'id': 1})"
1(["1<br>S:step2_L:0"])
click 1 callback "step: 'step2'<br>leadtime: 0<br>call: (#lt;function concatenate_id at 0x7faa06e21800#gt;, {'id': 2})"
2(["2<br>S:step3_L:0"])
click 2 callback "step: 'step3'<br>leadtime: 0<br>call: (#lt;function concatenate_id at 0x7faa06e21800#gt;, {'id': 3})"
3(["3<br>S:step4_L:0"])
click 3 callback "step: 'step4'<br>leadtime: 0<br>call: (#lt;function concatenate_id at 0x7faa06e21800#gt;, {'id': 4})"
4(["4<br>S:step5_L:0"])
click 4 callback "step: 'step5'<br>leadtime: 0<br>call: (#lt;function concatenate_id at 0x7faa06e21800#gt;, {'id': 5})"
0 --> 1
2 --> 3
3 --> 4
1 --> 4

Loading

We can see how the order of execution is respecting insertion order that we have changed.

---

If you want an approach to constructing your NetworkX graphs that informs this insertion order, then you can achieve this by something like (illustration only):

edges_append(edges, [node4, node5], arg_order=1)
edges_append(edges, [node2, node5], arg_order=0)

Perhaps we will do this in our future graph construction 🤷 but I am hoping that largely that making applications handle their input take us much of the way to this issue.

I can add an example configuration into the pp_demo_workflow repository that demonstrates NetworkX graph construction that ensures order in this way. Perhaps useful to others as a how-to, to NetworkX graph construction.

Some background:

NetworkX graphs Respect Insertion Order as its classes (like Graph, DiGraph, etc.) are internally implemented using Python dictionaries. Since Python 3.7, dictionaries officially preserve insertion order as part of the language specification. As a result, from NetworkX 2.6 onward, the default graph classes began to naturally preserve insertion order for nodes and edges without the use of additional classes for this purpose (OrderedDigraph etc.). This ordered NetworkX classes were actually removed in NetworkX 3.0 (FYI: networkx/networkx#4614).

[dmentipl]

Apologies @dmentipl. I think we are perhaps talking passed one another.

It's OK, I'm just trying to understand the approach. I think DAGRunner is a major step forward (especially compared to Paraflow).

Whilst most problems can be treated as graph problems, things become difficult to maintain, understand and support as it scales and changes over time. [...]

This is an important point. Trying to find the balance here. I like the explicit graphing style because it makes the workflow very, very explicit. But as you say, not everything can be treated this way and it can make things more complex. Moving complexity to the application layer makes sense in many cases.

NetworkX graphs Respect Insertion Order as its classes (like Graph, DiGraph, etc.) are internally implemented using Python dictionaries. Since Python 3.7, dictionaries officially preserve insertion order as part of the language specification. As a result, from NetworkX 2.6 onward, the default graph classes began to naturally preserve insertion order for nodes and edges without the use of additional classes for this purpose (OrderedDigraph etc.). This ordered NetworkX classes were actually removed in NetworkX 3.0 (FYI: networkx/networkx#4614).

Yes, I was digging into the networkx source to see how this was handled, and noticed that. It seems like an implementation detail to me.

---

Thanks very much for your explanations and clarifications. This has helped me to understand the design decisions. I have more questions but they can be left for an offline discussion.

[cpelley]

Yes, I was digging into the networkx source to see how this was handled, and noticed that. It seems like an implementation detail to me.

I agree that their mention of this is their docs isn't as strong a statement as the one they made when they deprecated the ordered variants of the four networkx graph classes.

docs:

NetworkX uses dicts to store the nodes and neighbors in a graph. So the reporting of nodes and edges for the base graph classes may not necessarily be consistent across versions and platforms; however, the reporting for CPython is consistent across platforms and versions after 3.6. - diff link

Deprecation message of the ordered classes:

OrderedGraph is deprecated and will be removed in version 3.0.
Use Graph instead, which guarantees order is preserved for
Python >= 3.7 - diff link

Opened an issue on this to hopefully make the contract explicit, as per deprecation messaging of the ordered variant classes.
networkx/networkx#8071