U /e@sddlmZddlZddlZddlZddlmZmZmZm Z m Z m Z ddl m Z mZddlZddlmZddlmZmZmZddlmZmZddlmZmZmZmZmZdd l m!Z!d d Z"Gd d d e Z#Gddde#Z$Gddde Z%dddZ&ddddZ'dS) annotationsN)HashableIterableKeysViewMappingMutableMappingSet)Anycast)config) clone_keyflattenis_dask_collection) keys_in_tasks reverse_dict) ensure_dict ensure_setimport_required key_split stringify) get_templatecspfdd}ddD}ddD}D]6\}}t|||D]}||||qRq4|S)z'Returns the dependencies between layerscs8D]\}}||kr|Sqtt|ddS)Nz not found)items RuntimeErrorrepr)keykvlayers7/tmp/pip-unpacked-wheel-dbjnr7gq/dask/highlevelgraph.py_find_layer_containing_keys z>compute_layer_dependencies.._find_layer_containing_keycSsh|]}|D]}|q qSr r ).0layerrr r r! sz-compute_layer_dependencies..cSsi|] }|tqSr )setr#rr r r! sz.compute_layer_dependencies..)valuesrrkeysadd)rr"all_keysretrrrr rr!compute_layer_dependenciess r.c@seZdZUdZded<ded<ddddd#d$d%d&Zdd'dd(d)d*d+Zed(d,d-d.d/d0d1Zd2d3Zd4d5Zd?d8d9Zd:d;ZdS)@LayeraxHigh level graph layer This abstract class establish a protocol for high level graph layers. The main motivation of a layer is to represent a collection of tasks symbolically in order to speedup a series of operations significantly. Ideally, a layer should stay in this symbolic state until execution but in practice some operations will force the layer to generate all its internal tasks. We say that the layer has been materialized. Most of the default implementations in this class will materialize the layer. It is up to derived classes to implement non-materializing implementations. Mapping[str, Any] | Nonercollection_annotationsN)rr1cCs4|pttdd|_|p,ttdd|_dS)aInitialize Layer object. Parameters ---------- annotations : Mapping[str, Any], optional By default, None. Annotations are metadata or soft constraints associated with tasks that dask schedulers may choose to respect: They signal intent without enforcing hard constraints. As such, they are primarily designed for use with the distributed scheduler. See the dask.annotate function for more information. collection_annotations : Mapping[str, Any], optional. By default, None. Experimental, intended to assist with visualizing the performance characteristics of Dask computations. These annotations are *not* passed to the distributed scheduler. rNr1)copyr getrr1)selfrr1r r r!__init__6s zLayer.__init__boolreturncCsdS)z/Return whether the layer is materialized or notTr r4r r r!is_materializedPszLayer.is_materializedr cCs|S)a?Return a set of all output keys Output keys are all keys in the layer that might be referenced by other layers. Classes overriding this implementation should not cause the layer to be materialized. Returns ------- keys: Set All output keys r*r9r r r!get_output_keysUszLayer.get_output_keysr&rz$tuple[Layer, Mapping[Hashable, set]])r* all_hlg_keysr8c st|tkr,fddDfSi}t}i}|}|r|}|||<|||<||D](}||krr|krr||||qrqBt|jd|fS)amRemove unnecessary tasks from the layer In other words, return a new Layer with only the tasks required to calculate `keys` and a map of external key dependencies. Examples -------- >>> inc = lambda x: x + 1 >>> add = lambda x, y: x + y >>> d = MaterializedLayer({'x': 1, 'y': (inc, 'x'), 'out': (add, 'x', 10)}) >>> _, deps = d.cull({'out'}, d.keys()) >>> deps {'out': {'x'}, 'x': set()} Returns ------- layer: Layer Culled layer deps: Map Map of external key dependencies csi|]}||qSr )get_dependenciesr'r=r4r r!r(szLayer.cull..r) lenr*r&r2popr>r+MaterializedLayerr) r4r*r=Zret_depsseenoutZworkrdr r?r!cullfs$   z Layer.cullr)rr=r8cCst|||gS)a:Get dependencies of `key` in the layer Parameters ---------- key: Hashable The key to find dependencies of all_hlg_keys: Iterable All keys in the high level graph. Returns ------- deps: set A set of dependencies r)r4rr=r r r!r>szLayer.get_dependencies)rr8csjtdt|jpi|pi}i}|D]<\}tr\fdd|D||<d||d<q(||<q(|S)aPacks Layer annotations for transmission to scheduler Callables annotations are fully expanded over Layer keys, while other values are simply transmitted as is Parameters ---------- annotations : Mapping[str, Any], optional A top-level annotations. Returns ------- packed_annotations : dict Packed annotations. zdict[str, Any]csi|]}t||qSr rr'rr r!r(sz?Layer.__dask_distributed_annotations_pack__..T__expanded_annotations__)r toolzmergerrcallable)r4rpackedar rIr!%__dask_distributed_annotations_pack__s z+Layer.__dask_distributed_annotations_pack__zMutableMapping[str, Any]Iterable[Hashable]None)rnew_annotationsr*r8cCs|dkr dSi}d}|D]Z\}}t|tkrPd|krP|}|d=|||<q|sfdd|D}d}t||||<q|D]\}}|||iq||dS)a Unpack a set of layer annotations across a set of keys, then merge those expanded annotations for the layer into an existing annotations mapping. This is not a simple shallow merge because some annotations like retries, priority, workers, etc need to be able to retain keys from different layers. Parameters ---------- annotations: MutableMapping[str, Any], input/output Already unpacked annotations, which are to be updated with the new unpacked annotations new_annotations: Mapping[str, Any], optional New annotations to be unpacked into `annotations` keys: Iterable All keys in the layer. NFrJcSsg|] }t|qSr rHr'r r r! szALayer.__dask_distributed_annotations_unpack__..T)rtypedictr2fromkeysupdater3)rrSr*expandedZkeys_stringifiedrOrrr r r!'__dask_distributed_annotations_unpack__s  z-Layer.__dask_distributed_annotations_unpack__ztuple[Layer, bool])r*seedbind_tor8c sddlm}fddi}d}|D]J\}}|krpt|}d|}|dk rprp|j||f}d}|||<q.t||fS)aCClone selected keys in the layer, as well as references to keys in other layers Parameters ---------- keys Keys to be replaced. This never includes keys not listed by :meth:`get_output_keys`. It must also include any keys that are outside of this layer that may be referenced by it. seed Common hashable used to alter the keys; see :func:`dask.base.clone_key` bind_to Optional key to bind the leaf nodes to. A leaf node here is one that does not reference any replaced keys; in other words it's a node where the replacement graph traversal stops; it may still have dependencies on non-replaced nodes. A bound node will not be computed until after ``bind_to`` has been computed. Returns ------- - New layer - True if the ``bind_to`` key was injected anywhere; False otherwise Notes ----- This method should be overridden by subclasses to avoid materializing the layer. r)chunkscst|}|tkrH|rHt|drH|dftfdd|ddDS|tkrbfdd|DS|tkrfdd |DSz|kr|WSWntk r|YSXd t|SdS) zhVariant of distributed.utils_comm.subs_multiple, which allows injecting bind_to rc3s|]}|VqdSNr r#i clone_valuer r! "sz3Layer.clone..clone_value..Ncsg|] }|qSr r r_rar r!rT$sz4Layer.clone..clone_value..csi|]\}}||qSr r r#rrrar r!r(&sz4Layer.clone..clone_value..F)rUtuplerMlistrVr TypeErrorr )otyprbZis_leafr*r[r r!rbs(  z Layer.clone..clone_valueFTN)Zdask.graph_manipulationr]rr bindrB) r4r*r[r\r]Zdsk_newboundrvaluer rkr!clones!     z Layer.clonezMapping[Hashable, Set]r )r=known_key_dependencies client_keysr8csddlmddlm}ddlm}m}ddlm}t |fdd D} | rb|| i} i}  D](\} } || dd \| | <}|rr|| | <qr| | rt j | nt }|D]4}|j|k rtd t|j|jkr|t|jqt|dd }| r"t | fd d| D}|}|fd d|D| D]<\} }|rTt|| dd }|dd|D||| <qTdd| D}Bfdd Dt||dS)aPack the layer for scheduler communication in Distributed This method should pack its current state and is called by the Client when communicating with the Scheduler. The Scheduler will then use .__dask_distributed_unpack__(data, ...) to unpack the state, materialize the layer, and merge it into the global task graph. The returned state must be compatible with Distributed's scheduler, which means it must obey the following: - Serializable by msgpack (notice, msgpack converts lists to tuples) - All remote data must be unpacked (see unpack_remotedata()) - All keys must be converted to strings now or when unpacking - All tasks must be serialized (see dumps_task()) The default implementation materialize the layer thus layers such as Blockwise and ShuffleLayer should implement a specialized pack and unpack function in order to avoid materialization. Parameters ---------- all_hlg_keys: Iterable[Hashable] All keys in the high level graph known_key_dependencies: Mapping[Hashable, Set] Already known dependencies client: distributed.Client The client calling this function. client_keys : Iterable[Hashable] List of keys requested by the client. Returns ------- state: Object serializable by msgpack Scheduler compatible state of the layer r)Future)CancelledError) subs_multipleunpack_remotedata) dumps_taskcs(i|] \}}t|r|kr||qSr ) isinstancere)rrrqr r!r(rs  z3Layer.__dask_distributed_pack__..T)Z byte_keysz;Inputs contain futures that were created by another client.r2csi|]\}}||qSr r re) alias_keysr r!r(sc3s&|]}|t|gddfVqdS)F)Zas_listNrGr')r=dskr r!rcsz2Layer.__dask_distributed_pack__..css|] }|jVqdSr^r)r#fr r r!rcscSs$i|]\}}t|dd|DqS)cSsh|] }t|qSr rH)r#depr r r!r%sz=Layer.__dask_distributed_pack__...rH)r#rdepsr r r!r(scs"i|]\}}t|t|dqS))Z exclusiverHre)merged_hlg_keysr r!r(s)rz dependencies)Zdistributed.clientrrZdistributed.utilsrsZdistributed.utils_commrtruZdistributed.workerrvrVrr&unionr)client ValueErrorrrfuturesrr*rXrrKZvalmap)r4r=rprrqrsrtrurvZfuture_aliasesZdsk2Zfut_depsrrZfutsZunpacked_futuresfuturer missing_keysrrEr )rrryr=rqrzrr!__dask_distributed_pack__@s^)            zLayer.__dask_distributed_pack__zMapping[str, Any]zMapping[str, set]rV)staterzrr8cCs|d|ddS)aaUnpack the state of a layer previously packed by __dask_distributed_pack__() This method is called by the scheduler in Distributed in order to unpack the state of a layer and merge it into its global task graph. The method can use `dsk` and `dependencies`, which are the already materialized state of the preceding layers in the high level graph. The layers of the high level graph are unpacked in topological order. See Layer.__dask_distributed_pack__() for packing detail. Parameters ---------- state: Any The state returned by Layer.__dask_distributed_pack__() dsk: Mapping, read-only The materialized low level graph of the already unpacked layers dependencies: Mapping, read-only The dependencies of each key in `dsk` Returns ------- unpacked-layer: dict layer_dsk: Mapping[str, Any] Materialized (stringified) graph of the layer layer_deps: Mapping[str, set] Dependencies of each key in `layer_dsk` rzr)rzr~r )clsrrzrr r r!__dask_distributed_unpack__s"z!Layer.__dask_distributed_unpack__cCstt|ffS)zBDefault serialization implementation, which materializes the Layer)rBrVr9r r r! __reduce__szLayer.__reduce__cCs"t||j}|j|j|S)z#Default shallow copy implementation)rU__new__ __class____dict__rX)r4objr r r!__copy__szLayer.__copy__r c Cs|dkrt|}nt|dr(t|j}n|jj}d}|jrn|jddkrn|jd}|rnddlm}||}t dj | |||| ||d S) NrnamerUdask.array.core.Arrayr]r)svgzhighlevelgraph_layer.html.j2)Z materialized shortname layer_indexhighlevelgraph_keyinforsvg_repr) rhasattrrr__name__r1r3Zdask.array.svgrrrenderr:layer_info_dict)r4rrrrrr]rr r r! _repr_html_s.     zLayer._repr_html_cCst|j|t|d}|jdk rP|jD]\}}tt |||<q4|j dk r|j D]"\}}|dkrdtt |||<qd|S)N) layer_typer:znumber of outputsr]) rUrr:r@r<rrhtmlescapestrr1)r4rrvalr r r!rs   zLayer.layer_info_dict)NN)N)N)rrr )r __module__ __qualname____doc____annotations__r5abcabstractmethodr:r<rFr>rP staticmethodrZror classmethodrrrrrr r r r!r/#s2 03Ks# r/csXeZdZdZdddfdd ZddZd d Zd d Zd dZddZ ddZ Z S)rBzFully materialized layer of `Layer` Parameters ---------- mapping: Mapping The mapping between keys and tasks, typically a dask graph. Nrmappingcstj|d||_dS)Nr)superr5r)r4rrrr r!r5szMaterializedLayer.__init__cCs ||jkSr^rr4rr r r! __contains__szMaterializedLayer.__contains__cCs |j|Sr^rrr r r! __getitem__szMaterializedLayer.__getitem__cCs t|jSr^)iterrr9r r r!__iter__"szMaterializedLayer.__iter__cCs t|jSr^)r@rr9r r r!__len__%szMaterializedLayer.__len__cCsdS)NTr r9r r r!r:(sz!MaterializedLayer.is_materializedcCs|Sr^r;r9r r r!r<+sz!MaterializedLayer.get_output_keys)N) rrrrr5rrrrr:r< __classcell__r r rr!rBsrBc@sfeZdZUdZded<ded<ded<ded <d ed <dNd dddddZeddZedOddZddZ ddddZ ddZ dddd Z d!dd"d#Z d dd$d%Zd&d'Zd(d)Zddd*d+Zed,d-Zd.d/Zed0d1ZdPd3d4Zd5d6Zd7dd8d9d:Zd;dd<d=d>Zd?d@ZdQdAdBddCdDdEZedddFdGdHZdIddJdKZdLdMZd S)RHighLevelGrapha Task graph composed of layers of dependent subgraphs This object encodes a Dask task graph that is composed of layers of dependent subgraphs, such as commonly occurs when building task graphs using high level collections like Dask array, bag, or dataframe. Typically each high level array, bag, or dataframe operation takes the task graphs of the input collections, merges them, and then adds one or more new layers of tasks for the new operation. These layers typically have at least as many tasks as there are partitions or chunks in the collection. The HighLevelGraph object stores the subgraphs for each operation separately in sub-graphs, and also stores the dependency structure between them. Parameters ---------- layers : Mapping[str, Mapping] The subgraph layers, keyed by a unique name dependencies : Mapping[str, set[str]] The set of layers on which each layer depends key_dependencies : Mapping[Hashable, set], optional Mapping (some) keys in the high level graph to their dependencies. If a key is missing, its dependencies will be calculated on-the-fly. Examples -------- Here is an idealized example that shows the internal state of a HighLevelGraph >>> import dask.dataframe as dd >>> df = dd.read_csv('myfile.*.csv') # doctest: +SKIP >>> df = df + 100 # doctest: +SKIP >>> df = df[df.name == 'Alice'] # doctest: +SKIP >>> graph = df.__dask_graph__() # doctest: +SKIP >>> graph.layers # doctest: +SKIP { 'read-csv': {('read-csv', 0): (pandas.read_csv, 'myfile.0.csv'), ('read-csv', 1): (pandas.read_csv, 'myfile.1.csv'), ('read-csv', 2): (pandas.read_csv, 'myfile.2.csv'), ('read-csv', 3): (pandas.read_csv, 'myfile.3.csv')}, 'add': {('add', 0): (operator.add, ('read-csv', 0), 100), ('add', 1): (operator.add, ('read-csv', 1), 100), ('add', 2): (operator.add, ('read-csv', 2), 100), ('add', 3): (operator.add, ('read-csv', 3), 100)} 'filter': {('filter', 0): (lambda part: part[part.name == 'Alice'], ('add', 0)), ('filter', 1): (lambda part: part[part.name == 'Alice'], ('add', 1)), ('filter', 2): (lambda part: part[part.name == 'Alice'], ('add', 2)), ('filter', 3): (lambda part: part[part.name == 'Alice'], ('add', 3))} } >>> graph.dependencies # doctest: +SKIP { 'read-csv': set(), 'add': {'read-csv'}, 'filter': {'add'} } See Also -------- HighLevelGraph.from_collections : typically used by developers to make new HighLevelGraphs zMapping[str, Layer]rzMapping[str, Set]rzdict[Hashable, Set]key_dependenciesrV_to_dictr&_all_external_keysNzMapping[str, Mapping]zdict[Hashable, Set] | None)rrrcCs(||_|p i|_dd|D|_dS)NcSs(i|] \}}|t|tr|nt|qSr )rwr/rBrer r r!r(sz+HighLevelGraph.__init__..)rrrr)r4rrrr r r!r5ws  zHighLevelGraph.__init__cCst|stt||}t|tr\t|jdd}|||<t|jdd}t | ||<n$t |}||||i}||h|t i}|||S)z4`from_collections` optimized for a single collectionTrx) rrhrU__dask_graph__rwrrrrr&__dask_layers___get_some_layer_name)rrr$ collectiongraphrr~rr r r!_from_collections   zHighLevelGraph._from_collectionr c Cst|dkr||||dS||i}|ti}tj|tdD]}t|r|}t|t r| |j | |j ||t| O<qt|}|||<|||t||<q>tt|q>|||S)acConstruct a HighLevelGraph from a new layer and a set of collections This constructs a HighLevelGraph in the common case where we have a single new layer and a set of old collections on which we want to depend. This pulls out the ``__dask_layers__()`` method of the collections if they exist, and adds them to the dependencies for this new layer. It also merges all of the layers from all of the dependent collections together into the new layers for this graph. Parameters ---------- name : str The name of the new layer layer : Mapping The graph layer itself dependencies : List of Dask collections A list of other dask collections (like arrays or dataframes) that have graphs themselves Examples -------- In typical usage we make a new task layer, and then pass that layer along with all dependent collections to this method. >>> def add(self, other): ... name = 'add-' + tokenize(self, other) ... layer = {(name, i): (add, input_key, other) ... for i, input_key in enumerate(self.__dask_keys__())} ... graph = HighLevelGraph.from_collections(name, layer, dependencies=[self]) ... return new_collection(name, graph) rdrr{)r@rr&rKuniqueidrrrwrrXrrrrr+rhrU) rrr$rrr~rrrr r r!from_collectionss"#      zHighLevelGraph.from_collectionsc Csz|j||WStk r$YnXz|j|d|WStttfk rTYnX|jD](}z||WStk rYq`Xq`t|dS)Nr)rKeyError IndexErrorrhr))r4rrEr r r!rszHighLevelGraph.__getitem__intr7cCstdd|jDS)Ncss|]}t|VqdSr^r@)r#r$r r r!rcsz)HighLevelGraph.__len__..)sumrr)r9r r r!rszHighLevelGraph.__len__cCs t|Sr^)rto_dictr9r r r!rszHighLevelGraph.__iter__cCs4z|jWStk r.t|}|_|YSXdS)zIEfficiently convert to plain dict. This method is faster than dict(self).N)rAttributeErrorr)r4rDr r r!rs zHighLevelGraph.to_dictrcCs |S)zGet all keys of all the layers. This will in many cases materialize layers, which makes it a relatively expensive operation. See :meth:`get_all_external_keys` for a faster alternative. )rr*r9r r r!r*szHighLevelGraph.keyscCsPz|jWStk rJt}|jD]}||q(||_|YSXdS)zGet all output keys of all layers This will in most cases _not_ materialize any layers, which makes it a relative cheap operation. Returns ------- keys: set A set of all external keys N)rrr&rr)rXr<)r4r*r$r r r!get_all_external_keyss z$HighLevelGraph.get_all_external_keyscCs |Sr^)rrr9r r r!rszHighLevelGraph.itemscCs |Sr^)rr)r9r r r!r)szHighLevelGraph.valuescCsT|}||j}|rN|jD](}||@D]}||||j|<q4q$|jS)zGet dependencies of all keys This will in most cases materialize all layers, which makes it an expensive operation. Returns ------- map: Mapping A map that maps each key to its dependencies )r*rrr)r>)r4r,rr$rr r r!get_all_dependenciess z#HighLevelGraph.get_all_dependenciescCs t|jSr^)rrr9r r r! dependents1szHighLevelGraph.dependentscCs&tt|jddt|jdd|jS)NTrx)rrrrrr2r9r r r!r25s   zHighLevelGraph.copycGsni}i}|D]V}t|tr4||j||jq t|trZ||t|<t|t|<q t|q |||Sr^) rwrrXrrrrr&rh)rZgraphsrrgr r r!rL<s     zHighLevelGraph.merge dask-hlg.svgcKs(ddlm}t|f|}|||||S)a> Visualize this dask high level graph. Requires ``graphviz`` to be installed. Parameters ---------- filename : str or None, optional The name of the file to write to disk. If the provided `filename` doesn't include an extension, '.png' will be used by default. If `filename` is None, no file will be written, and the graph is rendered in the Jupyter notebook only. format : {'png', 'pdf', 'dot', 'svg', 'jpeg', 'jpg'}, optional Format in which to write output file. Default is 'svg'. color : {None, 'layer_type'}, optional (default: None) Options to color nodes. - None, no colors. - layer_type, color nodes based on the layer type. **kwargs Additional keyword arguments to forward to ``to_graphviz``. Examples -------- >>> x.dask.visualize(filename='dask.svg') # doctest: +SKIP >>> x.dask.visualize(filename='dask.svg', color='layer_type') # doctest: +SKIP Returns ------- result : IPython.diplay.Image, IPython.display.SVG, or None See dask.dot.dot_graph for more information. See Also -------- dask.dot.dot_graph dask.base.visualize # low level variant r)graphviz_to_file)dask.dotr to_graphviz)r4filenameformatkwargsrrr r r! visualizeKs&   zHighLevelGraph.visualizec Csdd|jD}dd|jD}g}|jD].\}}|D]}|||q>|s2||q2g}t|dkr|}||||D]*} || d8<|| dkr|| qqf|S)aSort the layers in a high level graph topologically Parameters ---------- hlg : HighLevelGraph The high level graph's layers to sort Returns ------- sorted: list List of layer names sorted topologically cSsi|]\}}|t|qSr rrer r r!r(sz3HighLevelGraph._toposort_layers..cSsi|] }|gqSr r r'r r r!r(srrd)rrappendr@rA) r4ZdegreeZ reverse_depsreadyrrr}r-r$Zrdepr r r!_toposort_layersws"      zHighLevelGraph._toposort_layersr)r*r8csddlm}tt|}}i}i}tD]}j|}|| } | r4| | |\} } t} | D] } | | O} qr| | 8} || O}| ||<t ||st |t s|r4t|t| kr4|| q4t|fdd|D}t|||S)aReturn new HighLevelGraph with only the tasks required to calculate keys. In other words, remove unnecessary tasks from dask. Parameters ---------- keys iterable of keys or nested list of keys such as the output of ``__dask_keys__()`` Returns ------- hlg: HighLevelGraph Culled high level graph r) Blockwisecsi|]}|j|@qSr )r)r# layer_nameZret_layers_keysr4r r!r(sz'HighLevelGraph.cull..)Z dask.layersrr&rrreversedrr intersectionr<rFr)rwrBr:r@rXr*r)r4r*rZkeys_setZ all_ext_keys ret_layersZ ret_key_depsrr$Z output_keysZ culled_layerZ culled_depsZ external_depsrEret_dependenciesr rr!rFs<         zHighLevelGraph.cullz Iterable[str])rr8cCsXt|}i}i}|rN|}|j|||<|j|||<||||O}qt||S)aReturn a new HighLevelGraph with only the given layers and their dependencies. Internally, layers are not modified. This is a variant of :meth:`HighLevelGraph.cull` which is much faster and does not risk creating a collision between two layers with the same name and different content when two culled graphs are merged later on. Returns ------- hlg: HighLevelGraph Culled high level graph )r&rArrr*r)r4rZto_visitrrrr r r! cull_layerss zHighLevelGraph.cull_layersc Cs|jD]L\}}||jkr0tdt|d|D] }||jkr4tt|dq4q |jD]}t|dsbtqbt|j}|j }| }||krtdt |dt ||D]F}|j|||krtdt|dt|j|dt||qdS) Nz dependencies[z] not found in layersz not found in dependenciesrzincorrect dependencies keys z expected zincorrect dependencies[z]: ) rrrrrr)rAssertionErrorr.r*r&) r4rr~r}r$rZdep_key1Zdep_key2rr r r!validates,    *zHighLevelGraph.validaterQr0)rqrr8c s\g}fddDD]8}||jt|j|j||||dqd|iS)aPack the high level graph for Scheduler -> Worker communication The approach is to delegate the packaging to each layer in the high level graph by calling .__dask_distributed_pack__() and .__dask_distributed_annotations_pack__() on each layer. Parameters ---------- client : distributed.Client The client calling this function. client_keys : Iterable[Hashable] List of keys requested by the client. annotations : Mapping[str, Any], optional A top-level annotations. Returns ------- data: dict Packed high level graph layers c3s|]}j|VqdSr^r)r#rr9r r!rc1sz;HighLevelGraph.__dask_distributed_pack__..)rrrrr) rrrrUrrrrrP)r4rrqrrr$r r9r!rs"z(HighLevelGraph.__dask_distributed_pack__)hlgr8c Csddlm}i}i}i}|dD]}|ddkr>tj}tj}n&||d}t||d} | j}| j}||d||} || d| d D]\} } || t | B|| <q|||d | d q |||d S) aUnpack the high level graph for Scheduler -> Worker communication The approach is to delegate the unpackaging to each layer in the high level graph by calling ..._unpack__() and ..._annotations_unpack__() on each layer. Parameters ---------- hlg: dict Packed high level graph layers Returns ------- unpacked-graph: dict dsk: dict[str, Any] Materialized (stringified) graph of all nodes in the high level graph deps: dict[str, set] Dependencies of each key in `dsk` annotations: dict[str, Any] Annotations for `dsk` r)import_allowed_modulerrNrrrzr~r)rzr~r) Zdistributed.protocol.serializerr/rrZgetattrrXrr3r&r*) rrrzr~annor$Z unpack_stateZ unpack_annomodrZunpacked_layerrrr r r!rCs$    z*HighLevelGraph.__dask_distributed_unpack__rc Csxt|jdt|jd}|d|jjd|jjdtt|d7}t| D]\}}|d|d|d 7}qT|S) Nz with z layers. <.z object at z>  z.  ) rUrr@rrrhexr enumerater)r4Zrepresentationr`Zlayerkeyr r r!__repr__vs ,zHighLevelGraph.__repr__cCs.tdjt|j|j||jt|dS)Nzhighlevelgraph.html.j2)rUrZtoposortZlayer_dependenciesZ n_outputs) rrrUrrrrr@rr9r r r!r}s zHighLevelGraph._repr_html_)N)r )rN)N) rrrrrr5rrrrrrrr*rrr)rpropertyrr2rLrrrFrrrrrrrr r r r!r/sJ A  7   ,E&.2rBTc" Ksddlm}m} tdd} |p i}|p(i}|p0i}|p8i}|p@i}||d<d|d<d|d <||| j|||d } i} |jD]} t|j| | | <q~t | }t | }i}| d }|d krd dgddgddgddgddgddgddgddgd}|jD]} | | }| | i}|| |d}||kr4dnt d| | |||d}tt|j| j}d|ddd| | d}|j| j}|rV| ddkr|d| dd| d d!| d"d#| d$d% 7}| dd&krVd'd(i}| d)}|d*| d+d,| | d-d%t|d.tt|d/krJt|nd0d% 7}|d1t||d2t||d3t||d kr| |d}d4| |d5<|d6t||d7d8| j|f|q|jD]2\} }| | }|D]}| |}| ||qq|d krd9} d:}!|D].\}}|d5r2|!d;|dd<|d=7}!q2|!d>7}!| | i}|d1t|!|d2d?|d@dA| j| f|| S)BNr)labelrgraphvizaJDrawing dask graphs with the graphviz visualization engine requires the `graphviz` python library and the `graphviz` system library. Please either conda or pip install as follows: conda install python-graphviz # either conda install python -m pip install graphviz # or pip install and follow installation instructionsrankdirZboxshapeZ helveticaZfontname) graph_attr node_attr edge_attrcolorrz#CCC7F9Fz#F9CCC7z#FFD9F2z#D9F2FFz#D9FFE6z#DBDEE5)ZDataFrameIOLayerZ ShuffleLayerZSimpleShuffleLayerZArrayOverlayLayerZBroadcastJoinLayerrZBlockwiseLayerrB)cachezA r/rz Layer with z Tasks. rUrz Array Shape: z Data Type: Zdtypez Chunk Size: chunksizez Chunk Type: Z chunk_typerzdask.dataframe.core.DataFramezpandas.core.frame.DataFrameZpandascolumnszNumber of Partitions: Z npartitionsz DataFrame Type: Zdataframe_typez DataFrame Columns: (z[...]rZfontsizeZtooltipTrdZ fillcolorstyleZfilledZKeyzn<zz
Legend: Layer types
z
>Z20margin0)rrrrrXZDigraphrr@rminr)maxr3rrrUrreplacer1 setdefaultnoderZedge)"hgZdata_attributesZfunction_attributesrrrrrrrrrZn_tasksr$Z min_tasksZ max_tasksrrZ layer_colorsrattrsZ node_labelZ node_sizerZ node_tooltipsZlayer_caZdftypecolsZ node_colorr~r}dep_nameZ legend_titleZ legend_labelr r r!rs           4 L       rrr7c Cs:z|\}|WSttfk r4tt|YSXdS)zLSomehow get a unique name for a Layer from a non-HighLevelGraph dask mappingN)rrrrr)rrr r r!rs  r)NNrNNN)( __future__rrr2rcollections.abcrrrrrr typingr r ZtlzrKZdaskr Z dask.baser rrZ dask.corerrZ dask.utilsrrrrrZ dask.widgetsrr.r/rBrrrr r r r!s>    o ^