Execute a given graph. The graph is executed in topological order as defined by dask.order until all leaf nodes, i.e. nodes without any dependents, are reached. The returned dictionary contains the results of the leaf nodes. If keys are required that are not part of the graph, they
(
dsk: Iterable[GraphNode] | Mapping[KeyType, GraphNode],
cache: MutableMapping[KeyType, object] | None = None,
keys: Container[KeyType] | None = None,
)
| 1053 | |
| 1054 | |
| 1055 | def execute_graph( |
| 1056 | dsk: Iterable[GraphNode] | Mapping[KeyType, GraphNode], |
| 1057 | cache: MutableMapping[KeyType, object] | None = None, |
| 1058 | keys: Container[KeyType] | None = None, |
| 1059 | ) -> MutableMapping[KeyType, object]: |
| 1060 | """Execute a given graph. |
| 1061 | |
| 1062 | The graph is executed in topological order as defined by dask.order until |
| 1063 | all leaf nodes, i.e. nodes without any dependents, are reached. The returned |
| 1064 | dictionary contains the results of the leaf nodes. |
| 1065 | |
| 1066 | If keys are required that are not part of the graph, they can be provided in the `cache` argument. |
| 1067 | |
| 1068 | If `keys` is provided, the result will contain only values that are part of the `keys` set. |
| 1069 | |
| 1070 | """ |
| 1071 | if isinstance(dsk, (list, tuple, set, frozenset)): |
| 1072 | dsk = {t.key: t for t in dsk} |
| 1073 | else: |
| 1074 | assert isinstance(dsk, dict) |
| 1075 | |
| 1076 | refcount: defaultdict[KeyType, int] = defaultdict(int) |
| 1077 | for vals in DependenciesMapping(dsk).values(): |
| 1078 | for val in vals: |
| 1079 | refcount[val] += 1 |
| 1080 | |
| 1081 | cache = cache or {} |
| 1082 | from dask.order import order |
| 1083 | |
| 1084 | priorities = order(dsk) |
| 1085 | |
| 1086 | for key, node in sorted(dsk.items(), key=lambda it: priorities[it[0]]): |
| 1087 | cache[key] = node(cache) |
| 1088 | for dep in node.dependencies: |
| 1089 | refcount[dep] -= 1 |
| 1090 | if refcount[dep] == 0 and keys and dep not in keys: |
| 1091 | del cache[dep] |
| 1092 | |
| 1093 | return cache |
| 1094 | |
| 1095 | |
| 1096 | def fuse_linear_task_spec(dsk, keys): |