\c @`sddlmZmZmZddlmZddlmZddlm Z ddl m Z ddl Z ddl Z ddlZddlZddlZddlmZmZmZmZddlmZd d lmZmZmZmZmZmZmZm Z d d l!m"Z"d d l#m$Z$m%Z%m&Z'd d l(m)Z)d dl*m+Z+m,Z,d dl-m.Z.m/Z/m0Z0d=Z1dZ2de3fdYZ4dddZ6dZ7dZ8e9dZ:dZ;dZ<d Z=d!Z>d"Z?d#Z@d$ZAe+ZBeBjCeDeeEeFeeGeHdeHeIeJeHeKf eeBjCeLd%ZMeBjCed&ZNeBjCeOd'ZPeBjCeQeRfd(ZSeBjCe3d)ZTiZUejVZWd*ZXd+ZYeBjZd,d-Z[eBjZd.d/Z\eBjZd0d1Z]d2Z^ie/j_d36e/j_d46e/j_d56e0j&d66e0j&d76Z`ydd8lambZcWnedk rn"Xe`jeiecj&d96ecj&d:6d;jfZgddddd<ZhdS(>i(tabsolute_importtdivisiontprint_function(t OrderedDict(tpartial(tmd5(tgetitemN(tmergetgroupbytcurrytidentity(tComposei(tapplytlongtunicodetIteratort is_dataclasstdataclass_fieldstMappingtcPickle(t thread_state(tflattentquotetget(thash_buffer_hex(tDispatcht ensure_dict(tconfigtlocaltthreadedtDaskMethodsMixintis_dask_collectiontcomputetpersisttoptimizet visualizettokenizetnormalize_tokencC`s3y|jdk SWnttfk r.tSXdS(s.Returns ``True`` if ``x`` is a dask collectionN(t__dask_graph__tNonetAttributeErrort TypeErrortFalse(tx((s(lib/python2.7/site-packages/dask/base.pyRscB`s8eZdZdZddedZdZdZRS(s/A mixin adding standard dask collection methodstmydaskc K`st|d|d|d||S(sRender the computation of this object's task graph using graphviz. Requires ``graphviz`` to be installed. Parameters ---------- filename : str or None, optional The name (without an extension) of the file to write to disk. If `filename` is None, no file will be written, and we communicate with dot using only pipes. format : {'png', 'pdf', 'dot', 'svg', 'jpeg', 'jpg'}, optional Format in which to write output file. Default is 'png'. optimize_graph : bool, optional If True, the graph is optimized before rendering. Otherwise, the graph is displayed as is. Default is False. color: {None, 'order'}, optional Options to color nodes. Provide ``cmap=`` keyword for additional colormap **kwargs Additional keyword arguments to forward to ``to_graphviz``. Examples -------- >>> x.visualize(filename='dask.pdf') # doctest: +SKIP >>> x.visualize(filename='dask.pdf', color='order') # doctest: +SKIP Returns ------- result : IPython.diplay.Image, IPython.display.SVG, or None See dask.dot.dot_graph for more information. See Also -------- dask.base.visualize dask.dot.dot_graph Notes ----- For more information on optimization see here: https://docs.dask.org/en/latest/optimize.html tfilenametformattoptimize_graph(R#(tselfR-R.R/tkwargs((s(lib/python2.7/site-packages/dask/base.pyR#+s,cK`st|dt|\}|S(sPersist this dask collection into memory This turns a lazy Dask collection into a Dask collection with the same metadata, but now with the results fully computed or actively computing in the background. The action of function differs significantly depending on the active task scheduler. If the task scheduler supports asynchronous computing, such as is the case of the dask.distributed scheduler, then persist will return *immediately* and the return value's task graph will contain Dask Future objects. However if the task scheduler only supports blocking computation then the call to persist will *block* and the return value's task graph will contain concrete Python results. This function is particularly useful when using distributed systems, because the results will be kept in distributed memory, rather than returned to the local process as with compute. Parameters ---------- scheduler : string, optional Which scheduler to use like "threads", "synchronous" or "processes". If not provided, the default is to check the global settings first, and then fall back to the collection defaults. optimize_graph : bool, optional If True [default], the graph is optimized before computation. Otherwise the graph is run as is. This can be useful for debugging. **kwargs Extra keywords to forward to the scheduler function. Returns ------- New dask collections backed by in-memory data See Also -------- dask.base.persist ttraverse(R!R*(R0R1tresult((s(lib/python2.7/site-packages/dask/base.pyR!Zs'cK`st|dt|\}|S(sCompute this dask collection This turns a lazy Dask collection into its in-memory equivalent. For example a Dask.array turns into a :func:`numpy.array` and a Dask.dataframe turns into a Pandas dataframe. The entire dataset must fit into memory before calling this operation. Parameters ---------- scheduler : string, optional Which scheduler to use like "threads", "synchronous" or "processes". If not provided, the default is to check the global settings first, and then fall back to the collection defaults. optimize_graph : bool, optional If True [default], the graph is optimized before computation. Otherwise the graph is run as is. This can be useful for debugging. kwargs Extra keywords to forward to the scheduler function. See Also -------- dask.base.compute R2(R R*(R0R1R3((s(lib/python2.7/site-packages/dask/base.pyR s(N( t__name__t __module__t__doc__t __slots__R'R*R#R!R (((s(lib/python2.7/site-packages/dask/base.pyR's / *cK`sItd|d|d|}t|t|||}||||S(sqCompute a graph as if it were of type cls. Allows for applying the same optimizations and default scheduler.t schedulertclsR(t get_schedulertoptimization_functionR(R9tdsktkeysR8RR1tscheduletdsk2((s(lib/python2.7/site-packages/dask/base.pytcompute_as_if_collectionscK`s|S(N((R<R=R1((s(lib/python2.7/site-packages/dask/base.pyt dont_optimizescC`st|dtS(Nt__dask_optimize__(tgetattrRA(R+((s(lib/python2.7/site-packages/dask/base.pyR;sc `s|jddp!tjdg}|rtt|}d|jD}x*|D]"fd|jD}qVWttt g|jD]$\\}}|||^q}nt |\}}|S(sO Convert many collections into a single dask graph, after optimization t optimizationscS`s%i|]\}}t||qS((t_extract_graph_and_keys(t.0topttval((s(lib/python2.7/site-packages/dask/base.pys s c`s4i|]*\}\}}|||f|qS(((RFtkR<R=(RG(s(lib/python2.7/site-packages/dask/base.pys s N( tpopR'RRRR;titemsRtmapRRE(t collectionsR/R1RDtgroupsR<R=t_((RGs(lib/python2.7/site-packages/dask/base.pytcollections_to_dsks   =c`sddlmg|D]}|j^q}g|D]}|j^q6}tfd|Dr|j|}n t|}||fS(sGiven a list of dask vals, return a single graph and a list of keys such that ``get(dsk, keys)`` is equivalent to ``[v.compute() for v in vals]``.i(tHighLevelGraphc3`s|]}t|VqdS(N(t isinstance(RFtgraph(RQ(s(lib/python2.7/site-packages/dask/base.pys s(thighlevelgraphRQR&t __dask_keys__tanyR(tvalstvtgraphsR=RS((RQs(lib/python2.7/site-packages/dask/base.pyREs c`s|jdtgitjjfdtjjtg|D]}|^qaf<fd}|fS(sExtract collections in preparation for compute/persist/etc... Intended use is to find all collections in a set of (possibly nested) python objects, do something to them (compute, etc...), then repackage them in equivalent python objects. Parameters ---------- *args Any number of objects. If it is a dask collection, it's extracted and added to the list of collections returned. By default, python builtin collections are also traversed to look for dask collections (for more information see the ``traverse`` keyword). traverse : bool, optional If True (default), builtin python collections are traversed looking for any dask collections they might contain. Returns ------- collections : list A list of all dask collections contained in ``args`` repack : callable A function to call on the transformed collections to repackage them as they were in the original ``args``. R2c `st|rQt|}|krMttf|<j|n|Stjj}sut|}nt |t rt n t |}|t t tfkr|g|D]}|^qf}n|ttfkr%|g|jD]$\}}||g^qf}n_t|rt|dtgt|D]'}|jt||jg^qJff}n|S||<|S(N((RR$Rtlentappendtuuidtuuid4thexRRRRtlistttypettupletsettdictRRKRR RtnameRC(texprttokttskttyptiRIRXtf(t_unpackRMtcollections_tokent repack_dskR2(s(lib/python2.7/site-packages/dask/base.pyRks,   !(:  @ c`s)j}t||>> import dask.array as da >>> a = da.arange(10, chunks=2).sum() >>> b = da.arange(10, chunks=2).mean() >>> a2, b2 = optimize(a, b) >>> a2.compute() == a.compute() True >>> b2.compute() == b.compute() True N( RuRPRt__dask_postpersist__R'textendRRUR[( RtR1RMRrR<tat postpersistsR=trts((s(lib/python2.7/site-packages/dask/base.pyR""s'4  c O`s|jdt}|jdt}td||\}}|sF|Std|jddd|d|jdd}t|||}g|D]}|j^q} g|D]}|j^q} ||| |} |gt| | D]!\} \} }| | |^qS(s?Compute several dask collections at once. Parameters ---------- args : object Any number of objects. If it is a dask object, it's computed and the result is returned. By default, python builtin collections are also traversed to look for dask objects (for more information see the ``traverse`` keyword). Non-dask arguments are passed through unchanged. traverse : bool, optional By default dask traverses builtin python collections looking for dask objects passed to ``compute``. For large collections this can be expensive. If none of the arguments contain any dask objects, set ``traverse=False`` to avoid doing this traversal. scheduler : string, optional Which scheduler to use like "threads", "synchronous" or "processes". If not provided, the default is to check the global settings first, and then fall back to the collection defaults. optimize_graph : bool, optional If True [default], the optimizations for each collection are applied before computation. Otherwise the graph is run as is. This can be useful for debugging. kwargs Extra keywords to forward to the scheduler function. Examples -------- >>> import dask.array as da >>> a = da.arange(10, chunks=2).sum() >>> b = da.arange(10, chunks=2).mean() >>> compute(a, b) (45, 4.5) By default, dask objects inside python collections will also be computed: >>> compute({'a': a, 'b': b, 'c': 1}) # doctest: +SKIP ({'a': 45, 'b': 4.5, 'c': 1},) R2R/R8RMRN( RJRsRuR:R'RPRUt__dask_postcompute__tzip(RtR1R2R/RMRrR>R<R+R=t postcomputesRpRzRjRx((s(lib/python2.7/site-packages/dask/base.pyR Ys'c`sMddlm}|jdd}|jdt}g}xC|D];}t|tttfro|j|qA|j |qAWg|D]}t|t r|^q}g|D]}t |r|^q}t t |d|} x|D]} | j| qW|jd} | dkr!dd lm} dd lj} | | y|jd Wntk r{| jjnXttrdd lj} t| jntjdfd jD}fd |jD|d>> x.visualize(filename='dask.pdf') # doctest: +SKIP >>> x.visualize(filename='dask.pdf', color='order') # 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 Notes ----- For more information on optimization see here: https://docs.dask.org/en/latest/optimize.html i(t dot_graphR-R,R/tcolortorderi(RNtcmapc`s5i|]+\}}t|dt|qS(tbytes(t _colorizeRs(RFRIRX(Rtmx(s(lib/python2.7/site-packages/dask/base.pys s c`s7i|]-\}}i|d6t|d6|qS(Rtlabel(tstr(RFRIRX(to(s(lib/python2.7/site-packages/dask/base.pys s tfunction_attributescS`s&i|]\}}i|d6|qS(R((RFRIRX((s(lib/python2.7/site-packages/dask/base.pys s tdata_attributessUnknown value color=%s(tdask.dotRRJR*RRR_RaRbRwR[RRRcRPtupdateRRtmatplotlib.pyplottpyplottKeyErrortcmtRdBuRRCtmaxtvaluesRKtNotImplementedError(RtR1RR-R/targs2targtdskstargs3R<tdRRtplttcolors((RRRs(lib/python2.7/site-packages/dask/base.pyR#sB. (%     c`s|jdt}|jdt}td||\}}|sF|Std|jddd|}tj|ryddlm}Wnt k rqXy |}Wnt k rqX|j |kr|j |d||} || Snt |||} gg} } x[|D]S} tt| j}| j\}}| j|| j|||fqW|| | |} tt| | g| D].\}}}|fd|D|^q}||S( sR Persist multiple Dask collections into memory This turns lazy Dask collections into Dask collections with the same metadata, but now with their results fully computed or actively computing in the background. For example a lazy dask.array built up from many lazy calls will now be a dask.array of the same shape, dtype, chunks, etc., but now with all of those previously lazy tasks either computed in memory as many small :class:`numpy.array` (in the single-machine case) or asynchronously running in the background on a cluster (in the distributed case). This function operates differently if a ``dask.distributed.Client`` exists and is connected to a distributed scheduler. In this case this function will return as soon as the task graph has been submitted to the cluster, but before the computations have completed. Computations will continue asynchronously in the background. When using this function with the single machine scheduler it blocks until the computations have finished. When using Dask on a single machine you should ensure that the dataset fits entirely within memory. Examples -------- >>> df = dd.read_csv('/path/to/*.csv') # doctest: +SKIP >>> df = df[df.name == 'Alice'] # doctest: +SKIP >>> df['in-debt'] = df.balance < 0 # doctest: +SKIP >>> df = df.persist() # triggers computation # doctest: +SKIP >>> df.value().min() # future computations are now fast # doctest: +SKIP -10 >>> df.value().max() # doctest: +SKIP 100 >>> from dask import persist # use persist function on multiple collections >>> a, b = persist(a, b) # doctest: +SKIP Parameters ---------- *args: Dask collections scheduler : string, optional Which scheduler to use like "threads", "synchronous" or "processes". If not provided, the default is to check the global settings first, and then fall back to the collection defaults. traverse : bool, optional By default dask traverses builtin python collections looking for dask objects passed to ``persist``. For large collections this can be expensive. If none of the arguments contain any dask objects, set ``traverse=False`` to avoid doing this traversal. optimize_graph : bool, optional If True [default], the graph is optimized before computation. Otherwise the graph is run as is. This can be useful for debugging. **kwargs Extra keywords to forward to the scheduler function. Returns ------- New dask collections backed by in-memory data R2R/R8RMi(tdefault_clientc`si|]}||qS(((RFRI(R(s(lib/python2.7/site-packages/dask/base.pys Ms N(RJRsRuR:R'tinspecttismethodtdistributed.clientRt ImportErrort ValueErrorRR!RPR_RRURvRwR[RcR}(RtR1R2R/RMRrR>RtclientRpR<R=RyRxta_keystrebuildtstateRztksR{tresults2((Rs(lib/python2.7/site-packages/dask/base.pyR!s@<          ;cO`sA|r||f}nttttt|jjS(s Deterministic token >>> tokenize([1, 2, '3']) '7d6a880cd9ec03506eee6973ff551339' >>> tokenize('Hello') == tokenize('Hello') True (RRRaRLR%tencodet hexdigest(RtR1((s(lib/python2.7/site-packages/dask/base.pyR$Us cC`stt|jdtS(Ntkey(R%tsortedRKR(R((s(lib/python2.7/site-packages/dask/base.pytnormalize_dictiscC`s%t|jtt|jfS(N(R`R4R%R_RK(R((s(lib/python2.7/site-packages/dask/base.pytnormalize_ordered_dictnscC`stt|dtS(NR(R%RR(R{((s(lib/python2.7/site-packages/dask/base.pyt normalize_setsscC`s"t|jttt|fS(N(R`R4R_RLR%(tseq((s(lib/python2.7/site-packages/dask/base.pyt normalize_seqxscC`sHt|dd}|dk r%|St|r;t|StjjS(Nt__dask_tokenize__(RCR'tcallabletnormalize_functionR\R]R^(Rtmethod((s(lib/python2.7/site-packages/dask/base.pytnormalize_object}s cC`sy t|SWntk rt|}ttdkrtEttdkrx+ttdddD] }t|=qmWnWdQXn|t|<|Stk rt|SXdS(Nii(tfunction_cacheRt_normalize_functionRZtfunction_cache_lockR_R)(tfuncR3RI((s(lib/python2.7/site-packages/dask/base.pyRs      cC`sit|tr|j}nt|trut|dd}|rXt|f|jn|j}td|DSt|t rtd|j D}|j rtdt |j j D}nd}t|j||fSy)tj|dd}d|kr|SWntk r&nXy#ddl}|j|ddSWntk rdt|SXdS(Ntfirstcs`s|]}t|VqdS(N(R(RFRj((s(lib/python2.7/site-packages/dask/base.pys scs`s|]}t|VqdS(N(R%(RFRi((s(lib/python2.7/site-packages/dask/base.pys scs`s'|]\}}|t|fVqdS(N(R%(RFRIRX((s(lib/python2.7/site-packages/dask/base.pys stprotocolit__main__(RRR t_partialR RCR'treversedtfuncsRaRRttkeywordsRRKRRtpickletdumpst Exceptiont cloudpickleR(RRRRttkwsR3R((s(lib/python2.7/site-packages/dask/base.pyRs0 %      tpandascC`s|ddl}tj|jd}tj|jd}tj|jd}tj|jd}dS(NicS`s|jt|jgS(N(RdR%R(tind((s(lib/python2.7/site-packages/dask/base.pytnormalize_indexscS`s"t|jt|j|jgS(N(R%tcodest categoriestordered(tcat((s(lib/python2.7/site-packages/dask/base.pytnormalize_categoricals  cS`s2|j|jt|jjdjt|jgS(Ni(RdtdtypeR%t_datatblocksRtindex(R{((s(lib/python2.7/site-packages/dask/base.pytnormalize_seriess cS`sKg|jjD]}|j^q }||j|jg7}ttt|S(N(RRRtcolumnsRR_RLR%(tdftblocktdata((s(lib/python2.7/site-packages/dask/base.pytnormalize_dataframes"(RR%tregistertIndext CategoricaltSeriest DataFrame(tpdRRRR((s(lib/python2.7/site-packages/dask/base.pytregister_pandass  tnumpyc`sddltjjdtjjfd}tjjttjjttjjfd}dS(NicS`s|jst||jfSt|drt|ddrt|jdrw|jjj |jjjj }nd}|j t j j |j |j|j|j|fS|jjrwy^y.tdj|jjdddd }Wn)tk rtdj|j}nXWqttfk rsyttj|tj}Wqttk rotjj}qtXqXniy%t|jd d jd }WnAtt t!fk rt|j"jd d jd }nX||j|j|jfS( NtmodeR-tctypesit-tencodingsutf-8terrorst surrogatepassRtKti1(#tshapeRRthasattrRCR'tbaseRtget_as_parametertvalueR-tostpathtgetmtimetstridest hasobjectRtjointflatRtUnicodeDecodeErrorR)RRtHIGHEST_PROTOCOLRR\R]R^traveltviewt BufferErrorR(RRn(R+toffsetR((s(lib/python2.7/site-packages/dask/base.pytnormalize_arrays2 ! .   %+c`s(t|j|jdjfS(NR`(R`R4Rtndarray(R+(Rtnp(s(lib/python2.7/site-packages/dask/base.pytnormalize_matrixsc`sIy*|j}t||kr)d|SWntk rDt|SXdS(Nsnp.(R4RCR(R(R+Rd(R(s(lib/python2.7/site-packages/dask/base.pytnormalize_ufuncs    ( RR%RRtmatrixRtreprtgenerictufunc(RR((RRs(lib/python2.7/site-packages/dask/base.pytregister_numpys  "$tscipyc C`sddlj}d}xx|jdf|jdf|jdf|jdf|jdf|jdfgD](\}}tj |t |d |qgWtj |j d }dS(Nic`s)tjtfd|DfS(Nc3`s$|]}tt|VqdS(N(R%RC(RFR(R+(s(lib/python2.7/site-packages/dask/base.pys s(R`R4R(R+tattrs((R+s(lib/python2.7/site-packages/dask/base.pytnormalize_sparse_matrix sRtoffsetsRtindicestindptrt blocksizetrowtcoltrowsRcS`s%t|jtt|jfS(N(R`R4R%RRK(R+((s(lib/python2.7/site-packages/dask/base.pytnormalize_dok_matrixs(RRR(RRR R R(RR R R(RRR R(RRR R(RR R( t scipy.sparsetsparset dia_matrixt bsr_matrixt coo_matrixt csr_matrixt csc_matrixt lil_matrixR%RRt dok_matrix(tspRR9RR((s(lib/python2.7/site-packages/dask/base.pytregister_scipys     c`shd tfdtD}tt|dj}ddt||}d|S(s Convert (r, g, b) triple to "#RRGGBB" string For use with ``visualize(color=...)`` Examples -------- >>> _colorize((255, 255, 255)) '#FFFFFF' >>> _colorize((0, 32, 128)) '#002080' ic3`s1|]'\}}|dt|dVqdS(iiN(RZ(RFRiRX(tt(s(lib/python2.7/site-packages/dask/base.pys ,sit0it#(tsumt enumerateR^tinttupperRZ(RRith((Rs(lib/python2.7/site-packages/dask/base.pyRs "tsynct synchronousssingle-threadedtthreadst threading(tmultiprocessingt processesR&s The get= keyword has been removed. Please use the scheduler= keyword instead with the name of the desired scheduler like 'threads' or 'processes' x.compute(scheduler='single-threaded') x.compute(scheduler='threads') x.compute(scheduler='processes') or with a function that takes the graph and keys x.compute(scheduler=my_scheduler_function) or with a Dask client x.compute(scheduler=client) c`srttn|dk rt|r1|Sdt|jkr\t|dr\|jS|jt kr|t |jS|jdkrddl m }|jS|jdkrt d |qt d d j tt ntjd drtd tjd dStjddr;t tnttd trjddl m}|jjS|dk r}|jS|rg|D]}|dk r|^q}n|r|djtfd|Dst dnSdS(sK Get scheduler function There are various ways to specify the scheduler to use: 1. Passing in scheduler= parameters 2. Passing these into global confiuration 3. Using defaults of a dask collection This function centralizes the logic to determine the right scheduler to use from those many options tClientRsdask.distributedt distributedi(t get_clientR'R&s5Please install cloudpickle to use the '%s' scheduler.s!Expected one of [distributed, %s]s, R8R(t get_workerc3`s|]}|jkVqdS(N(t__dask_scheduler__(RFtc(R(s(lib/python2.7/site-packages/dask/base.pys ssCompute called on multiple collections with differing default schedulers. Please specify a scheduler=` parameter explicitly in compute or globally with `dask.config.set`.N(sdask.distributedR)(R'R&(R)t get_err_msgR'RR`R4RRtlowertnamed_schedulerstdistributed.workerR*RRRRR:RCRR*R+RR,tall(RR8RMR9R*R+R-((Rs(lib/python2.7/site-packages/dask/base.pyR:Ys@   $ "  ( (RRR R!R"R#R$R%(it __future__RRRRMRt functoolsRthashlibRtoperatorRRRRR%R\ttoolzRRR R ttoolz.functoolzR t compatibilityR R RRRRRRtcontextRtcoreRRRRothashingRtutilsRRtRRRt__all__RtobjectRR'R@RAR;RsRPRERuR"R R#R!R$R%RRtfloatRRR`tslicetcomplextEllipsisRcRRRbRRaR_RRRtLockRRRt register_lazyRRRRtget_syncR0tdaskR&tdask_multiprocessingRRtstripR.R:(((s(lib/python2.7/site-packages/dask/base.pyts     ": y     I 7 9 Z i  !   7