U /eJ@sddlmZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z m Z m Z mZddlmZmZmZddlmZddlmZddlZddlmZdd lmZmZmZmZmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.dd l/m0Z0dd l1m2Z2dd l3m4Z4dd l5m6Z6m7Z7m8Z8m9Z9m:Z:ddl;mZ>ddl?m@Z@mAZAmBZBmCZCmDZDddlEmFZFmGZGddlHmIZIddlJmKZKmLZLmMZMddlNmOZOddlPmQZQmRZRmSZSmTZTmUZUmVZVmWZWmXZXmYZYmZZZm[Z[m\Z\e8]de8dZ^dZ_e`deafddddZbddd Zcd!d"Zddd#d$Zedd%d&Zfd'd(Zgdd)e[ddd*fd+d,Zhd-d.Zid/d0ZjGd1d2d2Zkd3d4ZlGd5d6d6e6ZmGd7d8d8e6Zndd9d:Zoddd?Zqdd@dAZrdBdCZsdDdEZtdFdGZudHdIZvdJdKZwddLdMZxddNdOZydPdQZzdRdSZ{dTdUZ|ddVdWZ}GdXdYdYe Z~dZd[Zd\d]Zd^d_Zd`daZe_fdbdcZdddeZedffdgdhZddidjZdkdlZdmdnZddodpZdqdrZddsdtZdudvZdwdxZdydzZd{d|Zd}d~ZddZddZdS)) annotationsN) defaultdict)HashableIterableIteratorMapping)partialreducewraps)Random)urlopen) open_files) accumulatecomposecountcurryfirst frequenciesgroupbyjoinmerge merge_with partition_allpeekpluckreducebyremovesecondtaketopkuniquevalmap)config)chunkto_avro)DaskMethodsMixin dont_optimizenamed_schedulersreplace_name_in_keytokenize) blockwise) globalmethod)flattenget_dependenciesistaskquote reverse_dict)Delayedunpack_collections)HighLevelGraph)cullfuseinline)sizeof) applydigit ensure_bytes ensure_dictensure_unicodefuncnameinsert iter_chunks key_split parse_bytessystem_encodingtakes_multiple_argumentsZ processessyncZ__no__default__ no_resultcCsdS)NrFrGselfrGrG1/tmp/pip-unpacked-wheel-dbjnr7gq/dask/bag/core.pyLrK) __slots__ __reduce__TcCst|tkr&t|dkr&dd|DSt|s2|S|d|dd}}|sn|ttfkrn|d}t|ddiS|ftd d |DSdS) a Given a task, remove unnecessary calls to ``list`` and ``reify``. This traverses tasks and small lists. We choose not to traverse down lists of size >= 50 because it is unlikely that sequences this long contain other sequences in practice. Examples -------- >>> def inc(x): ... return x + 1 >>> task = (sum, (list, (map, inc, [1, 2, 3]))) >>> lazify_task(task) # doctest: +ELLIPSIS (, (, , [1, 2, 3])) 2cSsg|]}t|dqS)F lazify_task.0argrGrGrJ aszlazify_task..rNstartFcss|]}t|dVqdS)FNrPrRrGrGrJ iszlazify_task..)typelistlenr/reifyrQtuple)taskrWheadtailrGrGrJrQPsrQcCs tt|S)zq Remove unnecessary calls to ``list`` in tasks. See Also -------- dask.bag.core.lazify_task )r!rQdskrGrGrJlazifylsrccsl|dkrfddD}t|fddD}|t|t|dd|D] }|=q\S)zInline lists that are only used once. >>> d = {'b': (list, 'a'), ... 'c': (sum, 'b', 1)} >>> inline_singleton_lists(d, 'c') {'c': (, (, 'a'), 1)} Pairs nicely with lazify afterwards. Ncsi|]\}}|t|dqS))r^)r.rSkvrarGrJ sz*inline_singleton_lists..cs<h|]4\}}t|r|r|dtkrt|dkr|qS)rrV)r/rZr[rd) dependentsrGrJ s  z)inline_singleton_lists..F)Zinline_constants)itemsr1difference_updater-r7)rbkeys dependenciesZ inline_keysrerG)rhrbrJinline_singleton_listsws  rnc Ks^t|}t||\}}i}|dk r*||d<t|||p6g|f|\}}t|||}t|} | S)z Optimize a dask from a dask Bag.NZ rename_keys)r<r5r6rnrc) rbrlZ fuse_keysZrename_fused_keyskwargsdsk2rmdsk3dsk4Zdsk5rGrGrJoptimizes rsc Csp|b}t|tjrd}t}nd}t}d}|D]&}|r@||nd}|||q,|rb||W5QRXdS)N  FT) isinstanceio TextIOWrapperr=r;write)dataZ lazy_file last_endlinefendlineZensurestarteddrGrGrJ_to_textfiles_chunks  rinferFc s|dkr dnd} t|f|| ||jd|p,i} dtjfddt| D} tj| gd} t| j} |r| j f|d d | DS| SdS) alWrite dask Bag to disk, one filename per partition, one line per element. **Paths**: This will create one file for each partition in your bag. You can specify the filenames in a variety of ways. Use a globstring >>> b.to_textfiles('/path/to/data/*.json.gz') # doctest: +SKIP The * will be replaced by the increasing sequence 1, 2, ... :: /path/to/data/0.json.gz /path/to/data/1.json.gz Use a globstring and a ``name_function=`` keyword argument. The name_function function should expect an integer and produce a string. Strings produced by name_function must preserve the order of their respective partition indices. >>> from datetime import date, timedelta >>> def name(i): ... return str(date(2015, 1, 1) + i * timedelta(days=1)) >>> name(0) '2015-01-01' >>> name(15) '2015-01-16' >>> b.to_textfiles('/path/to/data/*.json.gz', name_function=name) # doctest: +SKIP :: /path/to/data/2015-01-01.json.gz /path/to/data/2015-01-02.json.gz ... You can also provide an explicit list of paths. >>> paths = ['/path/to/data/alice.json.gz', '/path/to/data/bob.json.gz', ...] # doctest: +SKIP >>> b.to_textfiles(paths) # doctest: +SKIP **Compression**: Filenames with extensions corresponding to known compression algorithms (gz, bz2) will be compressed accordingly. **Bag Contents**: The bag calling ``to_textfiles`` must be a bag of text strings. For example, a bag of dictionaries could be written to JSON text files by mapping ``json.dumps`` on to the bag first, and then calling ``to_textfiles`` : >>> b_dict.map(json.dumps).to_textfiles("/path/to/data/*.json") # doctest: +SKIP **Last endline**: By default the last line does not end with a newline character. Pass ``last_endline=True`` to invert the default. Nwbwt) compressionmodeencoding name_functionnumz to-textfiles-cs(i|] \}}|ftj|f|fqSrG)rname)rSir|br{rrGrJrgsz to_textfiles..rmcSsg|] }|jqSrG)path)rSr|rGrGrJrU sz to_textfiles..) r npartitionsuuiduuid4hex enumerater4from_collectionsrYcompute to_delayed)rrrrrrstorage_optionsr{rorfilesrbgraphoutrGrrJ to_textfiless,C  rcCsV|s|St|trt|}t|dtr@t|dts@t|}t|trRt|}|SNr)rvrrZrstrtoolzconcatresultsrGrGrJfinalizes   rcCs|dSrrGrrGrGrJ finalize_itemsrc@s8eZdZdZddZddZddZdd Zd d Zd S) StringAccessoraString processing functions Examples -------- >>> import dask.bag as db >>> b = db.from_sequence(['Alice Smith', 'Bob Jones', 'Charlie Smith']) >>> list(b.str.lower()) ['alice smith', 'bob jones', 'charlie smith'] >>> list(b.str.match('*Smith')) ['Alice Smith', 'Charlie Smith'] >>> list(b.str.split(' ')) [['Alice', 'Smith'], ['Bob', 'Jones'], ['Charlie', 'Smith']] cCs ||_dSN)_bag)rIbagrGrGrJ__init__0szStringAccessor.__init__cCstttt|ttSr)sortedsetdirrYrrHrGrGrJ__dir__3szStringAccessor.__dir__cOs|jtj|f||Sr)rmapoperator methodcaller)rIkeyargsrorGrGrJ_strmap6szStringAccessor._strmapcCsXzt||WStk rR|ttkrLtt|}t|t|j|YSYnXdSr) object__getattribute__AttributeErrorrrgetattr robust_wrapsrr)rIrfuncrGrGrJ __getattr__9s  zStringAccessor.__getattr__cCs ddlm}|jt||dS)aXFilter strings by those that match a pattern. Examples -------- >>> import dask.bag as db >>> b = db.from_sequence(['Alice Smith', 'Bob Jones', 'Charlie Smith']) >>> list(b.str.match('*Smith')) ['Alice Smith', 'Charlie Smith'] See Also -------- fnmatch.fnmatch r)fnmatch)pat)rrfilterr)rIpatternrrGrGrJmatchCs zStringAccessor.matchN) __name__ __module__ __qualname____doc__rrrrrrGrGrGrJrs  rcsfdd}|S)z-A weak version of wraps that only copies doc.cs j|_|Sr)r)wrappedwrapperrGrJ_Zszrobust_wraps.._rG)rrrGrrJrWs rc@seZdZd"ddZddZddZdd Zd d Zee d e d Z e e ZddZddZddddZe ddZeddZddZddZddZejZZZZd#d d!ZdS)$ItemNcCsR||_||_||_|p||_t|trN|j|jkrNtd|jdt|jdS)NzLayer z% not in the HighLevelGraph's layers: ) daskrr_layerrvr4Zlayers ValueErrorrZ)rIrbrlayerrGrGrJrbs z Item.__init__cCs|jSrrrHrGrGrJ__dask_graph__pszItem.__dask_graph__cCs|jgSrrrHrGrGrJ __dask_keys__sszItem.__dask_keys__cCs|jfSr)rrHrGrGrJ__dask_layers__vszItem.__dask_layers__cCs|jSrrrHrGrGrJ__dask_tokenize__yszItem.__dask_tokenize__ bag_optimizerZfalseycCstdfSNrG)rrHrGrGrJ__dask_postcompute__szItem.__dask_postcompute__cCs |jdfSr_rebuildrHrGrGrJ__dask_postpersist__szItem.__dask_postpersist__renamecCs |rt|j|n|j}t||Sr)r)rr)rIrbrrrGrGrJrsz Item._rebuildcCsTddlm}m}t||s,t|dr,||}t||s:tt|j|j| ddS)zfCreate bag item from a dask.delayed value. See ``dask.bag.from_delayed`` for details rr2delayedrr) dask.delayedr2rrvhasattrAssertionErrorrrrr)valuer2rrGrGrJ from_delayeds zItem.from_delayedcCs |j|jfSrrrrHrGrGrJ_argssz Item._argscCs|jSrrrHrGrGrJ __getstate__szItem.__getstate__cCs|\|_|_dSrrrIstaterGrGrJ __setstate__szItem.__setstate__cCsBdt|t||d}|||jfi}tj|||gd}t||S)N{}-{}r9r)formatr>r*rr4rr)rIrrrbrrGrGrJr9sz Item.applyTcCs:ddlm}|}|r(|||}||j||jdS)zConvert into a ``dask.delayed`` object. Parameters ---------- optimize_graph : bool, optional If True [default], the graph is optimized before converting into ``dask.delayed`` objects. rr2r)rr2r__dask_optimize__rrr)rIoptimize_graphr2rbrGrGrJrs zItem.to_delayed)N)T)rrrrrrrrr,rsr'r staticmethod DEFAULT_GET__dask_scheduler__rrrrpropertyrrrr9r&r__int__ __float__ __complex____bool__rrGrGrGrJras&   rc@s.eZdZdZddddddZdd Zd d d d ZddZddZe e de dZ e eZddZddZddddZddZeZeedZdd Zd!d"Zed#d$Zd%d&Zd'd(Zd)d*Zdsd+d,Zd-d.Zd/d0Z e!fd1d2Z"d3d4Z#e$e%dd5e&d6dd7fd8d9Z%e$e'dtdd?Z)dud@dAZ*dvdBdCZ+dwdDdEZ,de(dfdFdGZ-dxdHdIZ.dydJdKZ/dzdLdMZ0d{dNdOZ1d|dPdQZ2d}dRdSZ3dTdUZ4d~dWdXZ5ddYdZZ6dd[d\Z7d]d^Z8e!de!dfd_d`Z9ddbdcZ:dddeZ;dfdgZddmdnZ?ddodpZ@e!fdqdrZAdS)BagaParallel collection of Python objects Examples -------- Create Bag from sequence >>> import dask.bag as db >>> b = db.from_sequence(range(5)) >>> list(b.filter(lambda x: x % 2 == 0).map(lambda x: x * 10)) [0, 20, 40] Create Bag from filename or globstring of filenames >>> b = db.read_text('/path/to/mydata.*.json.gz').map(json.loads) # doctest: +SKIP Create manually (expert use) >>> dsk = {('x', 0): (range, 5), ... ('x', 1): (range, 5), ... ('x', 2): (range, 5)} >>> b = db.Bag(dsk, 'x', npartitions=3) >>> sorted(b.map(lambda x: x * 10)) [0, 0, 0, 10, 10, 10, 20, 20, 20, 30, 30, 30, 40, 40, 40] >>> int(b.fold(lambda x, y: x + y)) 30 rrint)rbrrcCs0t|tstj||gd}||_||_||_dS)Nr)rvr4rrrr)rIrbrrrGrGrJrs  z Bag.__init__cCs|jSrrrHrGrGrJrszBag.__dask_graph__zlist[Hashable])returncsfddtjDS)Ncsg|]}j|fqSrGrrSrrHrGrJrUsz%Bag.__dask_keys__..)rangerrHrGrHrJrszBag.__dask_keys__cCs|jfSrrrHrGrGrJrszBag.__dask_layers__cCs|jSrrrHrGrGrJrszBag.__dask_tokenize__rrcCstdfSr)rrHrGrGrJrszBag.__dask_postcompute__cCs |jdfSrrrHrGrGrJrszBag.__dask_postpersist__NrcCs(|j}|r|||}t||||jSr)rgetrYr)rIrbrrrGrGrJrs z Bag._rebuildcCsdt|j|jfS)Nzdask.bag<%s, npartitions=%d>)rArrrHrGrGrJ__str__sz Bag.__str__)fgetcOst||f||S)aApply a function elementwise across one or more bags. Note that all ``Bag`` arguments must be partitioned identically. Parameters ---------- func : callable *args, **kwargs : Bag, Item, or object Extra arguments and keyword arguments to pass to ``func`` *after* the calling bag instance. Non-Bag args/kwargs are broadcasted across all calls to ``func``. Notes ----- For calls with multiple `Bag` arguments, corresponding partitions should have the same length; if they do not, the call will error at compute time. Examples -------- >>> import dask.bag as db >>> b = db.from_sequence(range(5), npartitions=2) >>> b2 = db.from_sequence(range(5, 10), npartitions=2) Apply a function to all elements in a bag: >>> b.map(lambda x: x + 1).compute() [1, 2, 3, 4, 5] Apply a function with arguments from multiple bags: >>> from operator import add >>> b.map(add, b2).compute() [5, 7, 9, 11, 13] Non-bag arguments are broadcast across all calls to the mapped function: >>> b.map(add, 1).compute() [1, 2, 3, 4, 5] Keyword arguments are also supported, and have the same semantics as regular arguments: >>> def myadd(x, y=0): ... return x + y >>> b.map(myadd, y=b2).compute() [5, 7, 9, 11, 13] >>> b.map(myadd, y=1).compute() [1, 2, 3, 4, 5] Both arguments and keyword arguments can also be instances of ``dask.bag.Item``. Here we'll add the max value in the bag to each element: >>> b.map(myadd, b.max()).compute() [4, 5, 6, 7, 8] )bag_maprIrrrorGrGrJrs;zBag.mapc s|dttdfg}r>> def starmap(func, bag, **kwargs): ... return (func(*args, **kwargs) for args in bag) Parameters ---------- func : callable **kwargs : Item, Delayed, or object, optional Extra keyword arguments to pass to ``func``. These can either be normal objects, ``dask.bag.Item``, or ``dask.delayed.Delayed``. Examples -------- >>> import dask.bag as db >>> data = [(1, 2), (3, 4), (5, 6), (7, 8), (9, 10)] >>> b = db.from_sequence(data, npartitions=2) Apply a function to each argument tuple: >>> from operator import add >>> b.starmap(add).compute() [3, 7, 11, 15, 19] Apply a function to each argument tuple, with additional keyword arguments: >>> def myadd(x, y, z=0): ... return x + y + z >>> b.starmap(myadd, z=10).compute() [13, 17, 21, 25, 29] Keyword arguments can also be instances of ``dask.bag.Item`` or ``dask.delayed.Delayed``: >>> max_second = b.pluck(1).max() >>> max_second.compute() 10 >>> b.starmap(myadd, z=max_second).compute() [13, 17, 21, 25, 29] rstarmapcs(i|] }|fttj|fffqSrG)r\ starmap_chunkrrrrorrIrGrJrgpszBag.starmap..r) rr>r*unpack_scalar_dask_kwargsextendrrr4rrY)rIrrorm collectionsrbrrGrrJr=s-  z Bag.starmapcCs|j|j|jfSrrrrrHrGrGrJrwsz Bag._argscCs|jSrrrHrGrGrJr{szBag.__getstate__cCs|\|_|_|_dSrrrrGrGrJr~szBag.__setstate__csZdtdtfddtjD}tj|gd}t|jS)aFilter elements in collection by a predicate function. >>> def iseven(x): ... return x % 2 == 0 >>> import dask.bag as db >>> b = db.from_sequence(range(5)) >>> list(b.filter(iseven)) [0, 2, 4] zfilter--cs&i|]}|fttj|fffqSrG)r\rrrr predicaterIrGrJrgszBag.filter..rr>r*rrr4rrYrIr rbrrGr rJrs z Bag.filtercsdkrdksntdt|ts0t|}dt|tj|}fddttj|D}t j |gd}t |jS)aReturn elements from bag with probability of ``prob``. Parameters ---------- prob : float A float between 0 and 1, representing the probability that each element will be returned. random_state : int or random.Random, optional If an integer, will be used to seed a new ``random.Random`` object. If provided, results in deterministic sampling. Examples -------- >>> import dask.bag as db >>> b = db.from_sequence(range(5)) >>> list(b.random_sample(0.5, 43)) [0, 3, 4] >>> list(b.random_sample(0.5, 43)) [0, 3, 4] rrVz,prob must be a number in the interval [0, 1]zrandom-sample-%scs,i|]$\}}|fttj|f|ffqSrG)r\ random_sampler)rSrrrprobrIrGrJrgsz%Bag.random_sample..r) rrvr r*getstaterandom_state_data_pythonrziprr4rrY)rIr random_state state_datarbrrGrrJrs  zBag.random_samplecsZdtdtfddtjD}tj|gd}t|jS)zRemove elements in collection that match predicate. >>> def iseven(x): ... return x % 2 == 0 >>> import dask.bag as db >>> b = db.from_sequence(range(5)) >>> list(b.remove(iseven)) [1, 3] zremove-r cs&i|]}|fttj|fffqSrG)r\rrrr rGrJrgszBag.remove..rr r rGr rJrs z Bag.removecOst||f||S)akApply a function to every partition across one or more bags. Note that all ``Bag`` arguments must be partitioned identically. Parameters ---------- func : callable The function to be called on every partition. This function should expect an ``Iterator`` or ``Iterable`` for every partition and should return an ``Iterator`` or ``Iterable`` in return. *args, **kwargs : Bag, Item, Delayed, or object Arguments and keyword arguments to pass to ``func``. Partitions from this bag will be the first argument, and these will be passed *after*. Examples -------- >>> import dask.bag as db >>> b = db.from_sequence(range(1, 101), npartitions=10) >>> def div(nums, den=1): ... return [num / den for num in nums] Using a python object: >>> hi = b.max().compute() >>> hi 100 >>> b.map_partitions(div, den=hi).take(5) (0.01, 0.02, 0.03, 0.04, 0.05) Using an ``Item``: >>> b.map_partitions(div, den=b.max()).take(5) (0.01, 0.02, 0.03, 0.04, 0.05) Note that while both versions give the same output, the second forms a single graph, and then computes everything at once, and in some cases may be more efficient. )map_partitionsrrGrGrJrs)zBag.map_partitionscsdtttkr>fddtjD}nfddtjD}tj|gd}t|jS)akSelect item from all tuples/dicts in collection. >>> import dask.bag as db >>> b = db.from_sequence([{'name': 'Alice', 'credits': [1, 2, 3]}, ... {'name': 'Bob', 'credits': [10, 20]}]) >>> list(b.pluck('name')) ['Alice', 'Bob'] >>> list(b.pluck('credits').pluck(0)) [1, 10] zpluck-cs&i|]}|fttj|fffqSrGrZrrr)rrrIrGrJrgszBag.pluck..cs(i|] }|fttj|fffqSrGrrdefaultrrrIrGrJrgsr)r*r0 no_defaultrrr4rrY)rIrrrbrrGrrJrs z Bag.pluckcstfddt|DS)aTransform a bag of tuples to ``n`` bags of their elements. Examples -------- >>> import dask.bag as db >>> b = db.from_sequence([(i, i + 1, i + 2) for i in range(10)]) >>> first, second, third = b.unzip(3) >>> isinstance(first, db.Bag) True >>> first.compute() [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] Note that this is equivalent to: >>> first, second, third = (b.pluck(i) for i in range(3)) c3s|]}|VqdSr)rrrHrGrJrX szBag.unzip..r]r)rInrGrHrJunzipsz Bag.unziprTFc Ks t||||||f||d|S)N)rr{)r) rIrrrrrrr{rorGrGrJr"s  zBag.to_textfilesnull>c Kst|||||||||f | Srr$) rIfilenameZschemarrcodecZ sync_intervalmetadatarrorGrGrJr%:s  z Bag.to_avrocCs\|p|}|tk r2|jtt||dtt|||dSddlm}|j||||||dSdS)aParallelizable reduction Fold is like the builtin function ``reduce`` except that it works in parallel. Fold takes two binary operator functions, one to reduce each partition of our dataset and another to combine results between partitions 1. ``binop``: Binary operator to reduce within each partition 2. ``combine``: Binary operator to combine results from binop Sequentially this would look like the following: >>> intermediates = [reduce(binop, part) for part in partitions] # doctest: +SKIP >>> final = reduce(combine, intermediates) # doctest: +SKIP If only one function is given then it is used for both functions ``binop`` and ``combine`` as in the following example to compute the sum: >>> def add(x, y): ... return x + y >>> import dask.bag as db >>> b = db.from_sequence(range(5)) >>> b.fold(add).compute() 10 In full form we provide both binary operators as well as their default arguments >>> b.fold(binop=add, combine=add, initial=0).compute() 10 More complex binary operators are also doable >>> def add_to_set(acc, x): ... ''' Add new element x to set acc ''' ... return acc | set([x]) >>> b.fold(add_to_set, set.union, initial=set()).compute() {0, 1, 2, 3, 4} See Also -------- Bag.foldby initial) split_everyout_typer)r N)r reductionr_reduceZ tlz.curriedr )rIbinopcombiner$r%r&r rGrGrJfoldTs1  zBag.foldcCs2|jttt|ddt}|r.|jttdd}|S)zCount number of occurrences of each distinct element. >>> import dask.bag as db >>> b = db.from_sequence(['Alice', 'Bob', 'Alice']) >>> dict(b.frequencies()) # doctest: +SKIP {'Alice': 2, 'Bob', 1} rr&r%rT)rreverse)r'rmerge_frequenciesrr dictitemsrr)rIr%sortresultrGrGrJrszBag.frequenciescCsT|r.t|rt|rtt|}tt||d}n tt|}|j|t|tjt |ddS)aK largest elements in collection Optionally ordered by some key function >>> import dask.bag as db >>> b = db.from_sequence([10, 3, 5, 7, 11, 4]) >>> list(b.topk(2)) [11, 10] >>> list(b.topk(2, lambda x: -x)) [3, 4] rrr,) callablerDrr9rr'rrrr)rIrerr%rrGrGrJrs    zBag.topkcCsB|dkr tn tt|d}|dkr$tn tt|d}|j||tddS)a Distinct elements of collection Unordered without repeats. Parameters ---------- key: {callable,str} Defines uniqueness of items in bag by calling ``key`` on each item. If a string is passed ``key`` is considered to be ``lambda x: x[key]``. Examples -------- >>> import dask.bag as db >>> b = db.from_sequence(['Alice', 'Bob', 'Alice']) >>> sorted(b.distinct()) ['Alice', 'Bob'] >>> b = db.from_sequence([{'name': 'Alice'}, {'name': 'Bob'}, {'name': 'Alice'}]) >>> b.distinct(key=lambda x: x['name']).compute() [{'name': 'Alice'}, {'name': 'Bob'}] >>> b.distinct(key='name').compute() [{'name': 'Alice'}, {'name': 'Bob'}] Nrdistinct)r&r)chunk_distinctrmerge_distinctr'r)rIrrZaggrGrGrJr3sz Bag.distinctcs`|dkr d}|dkrj}t||}|p2td|jdkfddtjD}j}|pzt|d|} d } ||kr| t| } tt|t|D]*\} } t|fd d | Ddf|| | f<q| d}| | d7} qt|fd d t|Dd f|| d f<tj | |gd}|t krP| | d f|| <t || St || dSdS)aReduce collection with reduction operators. Parameters ---------- perpartition: function reduction to apply to each partition aggregate: function reduction to apply to the results of all partitions split_every: int (optional) Group partitions into groups of this size while performing reduction Defaults to 8 out_type: {Bag, Item} The out type of the result, Item if a single element, Bag if a list of elements. Defaults to Item. Examples -------- >>> import dask.bag as db >>> b = db.from_sequence(range(10)) >>> b.reduction(sum, sum).compute() 45 NF-part-rVcs$i|]}|ftj|ffqSrG)empty_safe_applyrr)ais_last perpartitionrIrGrJrgsz!Bag.reduction..z -aggregate-rcsg|] }|fqSrGrGrSjrrGrJrUsz!Bag.reduction..csg|] }|fqSrGrGr<r>rGrJrUsTr) rr*r>rrrrempty_safe_aggregater4rrpopr)rIr;Z aggregater%r&rtokenrbrefmtdepthcrindsrrG)r9rr:r;rIrJr'sF      z Bag.reductioncCs|jtt|dS)zSum all elementsr%)r'sumrIr%rGrGrJrG&szBag.sumcCs|jtt|dS)zMaximum elementrF)r'maxrHrGrGrJrI*szBag.maxcCs|jtt|dS)zMinimum elementrF)r'minrHrGrGrJrJ.szBag.mincCs|jtt|dS)zAre any of the elements truthy? Examples -------- >>> import dask.bag as db >>> bool_bag = db.from_sequence([True, True, False]) >>> bool_bag.any().compute() True rF)r'anyrHrGrGrJrK2s zBag.anycCs|jtt|dS)zAre all elements truthy? Examples -------- >>> import dask.bag as db >>> bool_bag = db.from_sequence([True, True, False]) >>> bool_bag.all().compute() False rF)r'allrHrGrGrJrL>s zBag.allcCs|jtt|dS)zCount the number of elements. Examples -------- >>> import dask.bag as db >>> numbers = db.from_sequence([1, 2, 3]) >>> numbers.count().compute() 3 rF)r'rrGrHrGrGrJrJs z Bag.countcCs dd}dd}|j||ddS)zArithmetic meancSs*d\}}|D]}||7}|d7}q ||fS)N)grrVrG)seqtotalrxrGrGrJ mean_chunkYs  zBag.mean..mean_chunkcSs$tt|\}}dt|t|S)Ng?)rZrrG)rOZtotalscountsrGrGrJmean_aggregate`sz Bag.mean..mean_aggregateFrF)r')rIrPrRrGrGrJmeanVszBag.meanrcCs|jtjttj|dddS)ZVarianceddofFrF)r'r#Z var_chunkrZ var_aggregaterIrUrGrGrJvarfs  zBag.varcCs|j|dtjS)zStandard deviationrT)rWr9mathsqrtrVrGrGrJstdlszBag.stdc Csdt||||}i}t|trb|jdkrT||j|d}t|f|d|<qd}t|nDt|t r||j|j }n&t|t r|}ndt |j }t||dkr|}t|jD]$}tt||||j|fff|||f<qtj|||gd}t ||||jS) a<Joins collection with another collection. Other collection must be one of the following: 1. An iterable. We recommend tuples over lists for internal performance reasons. 2. A delayed object, pointing to a tuple. This is recommended if the other collection is sizable and you're using the distributed scheduler. Dask is able to pass around data wrapped in delayed objects with greater sophistication. 3. A Bag with a single partition You might also consider Dask Dataframe, whose join operations are much more heavily optimized. Parameters ---------- other: Iterable, Delayed, Bag Other collection on which to join on_self: callable Function to call on elements in this collection to determine a match on_other: callable (defaults to on_self) Function to call on elements in the other collection to determine a match Examples -------- >>> import dask.bag as db >>> people = db.from_sequence(['Alice', 'Bob', 'Charlie']) >>> fruit = ['Apple', 'Apricot', 'Banana'] >>> list(people.join(fruit, lambda x: x[0])) [('Apple', 'Alice'), ('Apricot', 'Alice'), ('Banana', 'Bob')] zjoin-rVrz join-%s-otherzOMulti-bag joins are not implemented. We recommend Dask dataframe if appropriatezRJoined argument must be single-partition Bag, delayed object, or Iterable, got %sNr)r*rvrrupdaterrrZNotImplementedErrorr2_keyrrYZ _Bag__name TypeErrorrrrr4r) rIotherZon_selfZon_otherrrbmsgrrrGrGrJrps4#        "zBag.joincsnttstdtjj}fddt|D}tj|gd}t||S)z#Cartesian product between two bags.zproduct-c sDi|]<}tD].}||fttjj|fj|fffqqSrG)rrZ itertoolsproductrrSrr=mrr_rIrGrJrgs zBag.product..r) rvrrr*rrr4rrY)rIr_rrbrrGrdrJrbsz Bag.productc s|dkr d}|dkrj}t||}d||dkr@}tk rjfddtjD}nfddtjD}ttj|} d} j} | |krjt| } |tk rtt |t| D]8\} }t d| t j t tfd d |Dff|f|| | f<qnDtt |t| D]0\} }ttt|ffd d |Df|| | f<q"| d } | | d 7} qd |}|tk rtt d| t j t tfdd t| Dff|ff||df<n.tttt|ffdd t| Dff||df<tj||gd}t||d S)a Combined reduction and groupby. Foldby provides a combined groupby and reduce for efficient parallel split-apply-combine tasks. The computation >>> b.foldby(key, binop, init) # doctest: +SKIP is equivalent to the following: >>> def reduction(group): # doctest: +SKIP ... return reduce(binop, group, init) # doctest: +SKIP >>> b.groupby(key).map(lambda (k, v): (k, reduction(v)))# doctest: +SKIP But uses minimal communication and so is *much* faster. >>> import dask.bag as db >>> b = db.from_sequence(range(10)) >>> iseven = lambda x: x % 2 == 0 >>> add = lambda x, y: x + y >>> dict(b.foldby(iseven, add)) {True: 20, False: 25} **Key Function** The key function determines how to group the elements in your bag. In the common case where your bag holds dictionaries then the key function often gets out one of those elements. >>> def key(x): ... return x['name'] This case is so common that it is special cased, and if you provide a key that is not a callable function then dask.bag will turn it into one automatically. The following are equivalent: >>> b.foldby(lambda x: x['name'], ...) # doctest: +SKIP >>> b.foldby('name', ...) # doctest: +SKIP **Binops** It can be tricky to construct the right binary operators to perform analytic queries. The ``foldby`` method accepts two binary operators, ``binop`` and ``combine``. Binary operators two inputs and output must have the same type. Binop takes a running total and a new element and produces a new total: >>> def binop(total, x): ... return total + x['amount'] Combine takes two totals and combines them: >>> def combine(total1, total2): ... return total1 + total2 Each of these binary operators may have a default first value for total, before any other value is seen. For addition binary operators like above this is often ``0`` or the identity element for your operation. **split_every** Group partitions into groups of this size while performing reduction. Defaults to 8. >>> b.foldby('name', binop, 0, combine, 0) # doctest: +SKIP Examples -------- We can compute the maximum of some ``(key, value)`` pairs, grouped by the ``key``. (You might be better off converting the ``Bag`` to a ``dask.dataframe`` and using its groupby). >>> import random >>> import dask.bag as db >>> tokens = list('abcdefg') >>> values = range(10000) >>> a = [(random.choice(tokens), random.choice(values)) ... for _ in range(100)] >>> a[:2] # doctest: +SKIP [('g', 676), ('a', 871)] >>> a = db.from_sequence(a) >>> def binop(t, x): ... return max((t, x), key=lambda x: x[1]) >>> a.foldby(lambda x: x[0], binop).compute() # doctest: +SKIP [('g', ('g', 984)), ('a', ('a', 871)), ('b', ('b', 999)), ('c', ('c', 765)), ('f', ('f', 955)), ('e', ('e', 991)), ('d', ('d', 854))] See Also -------- toolz.reduceby pyspark.combineByKey Nr6Fz foldby-a-cs&i|]}|ftj|ffqSrGrrr)r9r)r$rrIrGrJrgCszBag.foldby..cs$i|]}|ftj|ffqSrGrfr)r9r)rrIrGrJrgHsrcsg|] }|fqSrGrGr<r>rGrJrUYszBag.foldby..csg|] }|fqSrGrGr<r>rGrJrUasrVz foldby-b-csg|] }|fqSrGrGr<r>rGrJrUpscsg|] }|fqSrGrGr<r>rGrJrUwsr)rr*rrrr#Zfoldby_combine2rrrrrrrr/rr r4rrY)rIrr)r$r*Zcombine_initialr%rArbZcombine2rCrerDrrEerrG)r9rr)r$rrIrJfoldbysht         z Bag.foldbyrVc s|dkr|j}||jkr*td|j|t|||}d|}|dkrd|i}t|D] }tt||j|fff||f<qZtj fddt|Df} t || |f||df<n|dft ||jdf|fi}t j |||gd } t | |d} |rt| S| Sd S) a&Take the first k elements. Parameters ---------- k : int The number of elements to return npartitions : int, optional Elements are only taken from the first ``npartitions``, with a default of 1. If there are fewer than ``k`` rows in the first ``npartitions`` a warning will be raised and any found rows returned. Pass -1 to use all partitions. compute : bool, optional Whether to compute the result, default is True. warn : bool, optional Whether to warn if the number of elements returned is less than requested, default is True. >>> import dask.bag as db >>> b = db.from_sequence(range(1_000)) >>> b.take(3) (0, 1, 2) z$only {} partitions, take received {}ztake-rVz take-partial-csg|] }|fqSrGrGrZname_prGrJrUszBag.take..rrN)rrrr*rrZrrrr safe_taker4rrr]r) rIrerrwarnrArrbrrrrrGrjrJr}s0     zBag.takecsJdtfddtjD}tj|gd}t|jS)zConcatenate nested lists into one long list. >>> import dask.bag as db >>> b = db.from_sequence([[1], [2, 3]]) >>> list(b) [[1], [2, 3]] >>> list(b.flatten()) [1, 2, 3] zflatten-cs&i|]}|fttjj|fffqSrG)rZrrrrrrIrGrJrgszBag.flatten..r)r*rrr4rrY)rIrbrrGrmrJr-s  z Bag.flattencCs t|Sr)iterrrHrGrGrJ__iter__sz Bag.__iter__cCs|dk rtd|dkr$tdd}|dkrFtdddkrBd}nd}|dkr^t||||dS|dkrtt|||d Sd }t|dS) aGroup collection by key function This requires a full dataset read, serialization and shuffle. This is expensive. If possible you should use ``foldby``. Parameters ---------- grouper: function Function on which to group elements shuffle: str Either 'disk' for an on-disk shuffle or 'tasks' to use the task scheduling framework. Use 'disk' if you are on a single machine and 'tasks' if you are on a distributed cluster. npartitions: int If using the disk-based shuffle, the number of output partitions blocksize: int If using the disk-based shuffle, the size of shuffle blocks (bytes) max_branch: int If using the task-based shuffle, the amount of splitting each partition undergoes. Increase this for fewer copies but more scheduler overhead. Examples -------- >>> import dask.bag as db >>> b = db.from_sequence(range(10)) >>> iseven = lambda x: x % 2 == 0 >>> dict(b.groupby(iseven)) # doctest: +SKIP {True: [0, 2, 4, 6, 8], False: [1, 3, 5, 7, 9]} See Also -------- Bag.foldby Nz.The method= keyword has been moved to shuffle=shuffleZ scheduler)zdask.distributedZ distributedZtasksZdisk)r blocksize) max_branchz!Shuffle must be 'disk' or 'tasks') Exceptionr"r groupby_disk groupby_tasksr\)rIgroupermethodrrrrsrqr`rGrGrJrs&+ z Bag.groupbyc Csddl}ddlm}|dkrR|jddd}t|dkr>td|jt||d}n&|dk rdtdn|jj ||d }t|j }|j }| t||} |r|| j| } n| j} dg|jd} || | j|| S) aCreate Dask Dataframe from a Dask Bag. Bag should contain tuples, dict records, or scalars. Index will not be particularly meaningful. Use ``reindex`` afterwards if necessary. Parameters ---------- meta : pd.DataFrame, dict, iterable, optional An empty ``pd.DataFrame`` that matches the dtypes and column names of the output. This metadata is necessary for many algorithms in dask dataframe to work. For ease of use, some alternative inputs are also available. Instead of a ``DataFrame``, a ``dict`` of ``{name: dtype}`` or iterable of ``(name, dtype)`` can be provided. If not provided or a list, a single element from the first partition will be computed, triggering a potentially expensive call to ``compute``. This may lead to unexpected results, so providing ``meta`` is recommended. For more information, see ``dask.dataframe.utils.make_meta``. columns : sequence, optional Column names to use. If the passed data do not have names associated with them, this argument provides names for the columns. Otherwise this argument indicates the order of the columns in the result (any names not found in the data will become all-NA columns). Note that if ``meta`` is provided, column names will be taken from there and this parameter is invalid. optimize_graph : bool, optional If True [default], the graph is optimized before converting into :class:`dask.dataframe.DataFrame`. Examples -------- >>> import dask.bag as db >>> b = db.from_sequence([{'name': 'Alice', 'balance': 100}, ... {'name': 'Bob', 'balance': 200}, ... {'name': 'Charlie', 'balance': 300}], ... npartitions=2) >>> df = b.to_dataframe() >>> df.compute() name balance 0 Alice 100 1 Bob 200 0 Charlie 300 rNrVF)rlzm`dask.bag.Bag.to_dataframe` failed to properly infer metadata, please pass in metadata via the `meta` keywordcolumnsz'Can't specify both `meta` and `columns`)Z parent_meta)pandasZdask.dataframeZ dataframerr[r DataFramerZutilsZ make_metarzdtypesZto_dictr to_dataframerrrrr) rImetarzrpdddr_colsr~dfsrbZ divisionsrGrGrJrs(0     zBag.to_dataframecs`ddlm|}||j|rJ||dtjddfdd|DS)aOConvert into a list of ``dask.delayed`` objects, one per partition. Parameters ---------- optimize_graph : bool, optional If True [default], the graph is optimized before converting into ``dask.delayed`` objects. See Also -------- dask.bag.from_delayed rrzdelayed-rGrcsg|]}|dqS)rrG)rSrer2rbrrGrJrUmsz"Bag.to_delayed..)rr2rrrrr4r)rIrrlrGrrJrWs  zBag.to_delayedcCsHt|dk |dk gdkr td|dk r2t||S|dk rDt||SdS)a Repartition Bag across new divisions. Parameters ---------- npartitions : int, optional Number of partitions of output. partition_size : int or string, optional Max number of bytes of memory for each partition. Use numbers or strings like 5MB. .. warning:: This keyword argument triggers computation to determine the memory size of each partition, which may be expensive. Notes ----- Exactly one of ``npartitions`` or ``partition_size`` should be specified. A ``ValueError`` will be raised when that is not the case. Examples -------- >>> b.repartition(5) # set to have 5 partitions # doctest: +SKIP NrVzRPlease provide exactly one ``npartitions`` or ``partition_size`` keyword arguments)rGrrepartition_npartitionsrepartition_size)rIrpartition_sizerGrGrJ repartitionos zBag.repartitionc Cst|||}t|}|d|}|d|}|d|}|dft||jdf|df|dft|dff|dft|dffi}td|jD]N} t||j| f|| dff||| f<t|| ff||| f<t|| ff||| f<qtj |||gd} t | ||jS)a{Repeatedly apply binary function to a sequence, accumulating results. This assumes that the bag is ordered. While this is typically the case not all Dask.bag functions preserve this property. Examples -------- >>> import dask.bag as db >>> from operator import add >>> b = db.from_sequence([1, 2, 3, 4, 5], npartitions=2) >>> b.accumulate(add).compute() [1, 3, 6, 10, 15] Accumulate also takes an optional argument that will be used as the first value. >>> b.accumulate(add, initial=-1).compute() [-1, 0, 2, 5, 9, 14] r7z-first-z-second-rTrVr) r*r>accumulate_partrrrrrr4rr) rIr)r$rAZ binop_namer9rrDrbrrrGrGrJrs$   "zBag.accumulate)N)NNrrNT)NF)NN)N)N)N)N)N)N)N)r)r)N)rVTT)NNrpNN)NNT)T)NN)Brrrrrrrrrr,rsr'rrrrrrrr__repr__rrrrrrrrrrrrrrrr rrCr%rr+rrr3r'rGrIrJrKrLrrSrWrZrrbrhrr-rorrrrrrGrGrGrJrs =:  #+    C    F       E 9 7 > O  "rcCsX|tkrtt||}ntt|||d}|rD||r<|dng|fS|dd|dfS)Nr#rirV)rrZr)r)rMr$Zis_firstresrGrGrJrs rrpc Cs`t||D]P}t||}tt}|D]"\}} |tt||| q(|j|ddq |S)z:Partition a bag along a grouper, store partitions on disk.T)fsync) rrrrZrjabshashrappend) rwsequencerpZ nelementsblockrZd2rerfrGrGrJ partitions rcCs t||j|dd}t|S)z7Collect partitions from disk and yield k,v group pairs.F)lock)rrrZrj)rwgroupr barrier_tokenrrGrGrJcollectsrcst|}|r&|s&ttt||}|dkrX|dkrXt|dkrHd}ntt|d}tt||}dt||t|dkrfddt|D}n dfgi}t|t|S)a"Create a dask Bag from Python sequence. This sequence should be relatively small in memory. Dask Bag works best when it handles loading your data itself. Commonly we load a sequence of filenames into a Bag and then use ``.map`` to open them. Parameters ---------- seq: Iterable A sequence of elements to put into the dask partition_size: int (optional) The length of each partition npartitions: int (optional) The number of desired partitions It is best to provide either ``partition_size`` or ``npartitions`` (though not both.) Examples -------- >>> import dask.bag as db >>> b = db.from_sequence(['Alice', 'Bob', 'Chuck'], partition_size=2) See Also -------- read_text: Create bag from text files NdrVzfrom_sequence-rcsi|]\}}|ft|qSrG)rZ)rSrpartrrGrJrgsz!from_sequence..) rZrrXceilr[rr*rr)rMrrpartsrrGrrJ from_sequences   rcCsXt|tr|g}dtj}i}t|D]\}}tt|ff|||f<q*t||t |S)aJCreate a dask Bag from a url. Examples -------- >>> a = from_url('http://raw.githubusercontent.com/dask/dask/main/README.rst') >>> a.npartitions 1 >>> a.take(8) # doctest: +SKIP (b'Dask\n', b'====\n', b'\n', b'|Build Status| |Coverage| |Doc Status| |Discourse| |Version Status| |NumFOCUS|\n', b'\n', b'Dask is a flexible parallel computing library for analytics. See\n', b'documentation_ for more information.\n', b'\n') >>> b = from_url(['http://github.com', 'http://google.com']) >>> b.npartitions 2 z from_url-) rvrrrrrrZr rr[)urlsrrbrurGrGrJfrom_urls rcCs t|S)zSA pickleable version of dict.items >>> dictitems({'x': 1}) [('x', 1)] )rZrj)rrGrGrJr/!sr/csJdt|tdfdd|D}tj||d}t|t|S)zConcatenate many bags together, unioning all elements. >>> import dask.bag as db >>> a = db.from_sequence([1, 2, 3]) >>> b = db.from_sequence([4, 5, 6]) >>> c = db.concat([a, b]) >>> list(c) [1, 2, 3, 4, 5, 6] zconcat-rcs(i|] }|D]}tf|qqSrG)rnext)rSrrcounterrrGrJrg7s   zconcat..r)r*rarr4rrr[bagsrbrrGrrJr*s  rcCs:t|trt|}t|r6t|dtr6ttt|}|Sr)rvrrZr[r)rMrGrGrJr\<s  r\csddlmmt|r |g}fdd|D}dt|fddtt|D}dd|D}tt||}t j ||d}t |t|S) aWCreate bag from many dask Delayed objects. These objects will become the partitions of the resulting Bag. They should evaluate to a ``list`` or some other concrete sequence. Parameters ---------- values: list of delayed values An iterable of dask Delayed objects. Each evaluating to a list. Returns ------- Bag Examples -------- >>> x, y, z = [delayed(load_sequence_from_file)(fn) ... for fn in filenames] # doctest: +SKIP >>> b = from_delayed([x, y, z]) # doctest: +SKIP See also -------- dask.delayed rrcs,g|]$}t|s$t|dr$|n|qSr)rvrrSrfrrGrJrUasz from_delayed..zbag-from-delayed-csg|] }|fqSrGrGrrrGrJrUgscSsg|]}t|jfqSrG)r\rrrGrGrJrUhsr) rr2rrvr*rr[dictrr4rr)valuesnamesZvalues2rbrrG)r2rrrJrDs   rcCs.|dk rt|sttj|d}tt||dSNr)r2rr#getitemrZr )rMrrGrGrJr4osr4cCstt||dSr)r4rr)seqsrrGrGrJr5usr5cCszt|trt|}|siS|d|dd}}|s8|Stt}|||D]&}|D]\}}|||7<qZqN|S)NrrV)rvrrZrrr[rj)rrrestrrrerfrGrGrJr.ys  r.cs~|d|fttt|td}fdd|D}|dkrr|d\}}tt|ff||f<t||S)zNumbers from zero to n Examples -------- >>> import dask.bag as db >>> b = db.range(5, npartitions=2) >>> list(b) [0, 1, 2, 3, 4] zrange-%d-npartitions-%drc s.i|]&\}}|ftt|t|ffqSrG)r\rrJrcrrsizerGrJrgszbag_range..ri)rZrrrr\r)rrZijsrbrr=rGrrJ bag_ranges    rcsddjtfddDs$tdtfddtD}tj|d}t|S)a%Partition-wise bag zip All passed bags must have the same number of partitions. NOTE: corresponding partitions should have the same length; if they do not, the "extra" elements from the longer partition(s) will be dropped. If you have this case chances are that what you really need is a data alignment mechanism like pandas's, and not a missing value filler like zip_longest. Examples -------- Correct usage: >>> import dask.bag as db >>> evens = db.from_sequence(range(0, 10, 2), partition_size=4) >>> odds = db.from_sequence(range(1, 10, 2), partition_size=4) >>> pairs = db.zip(evens, odds) >>> list(pairs) [(0, 1), (2, 3), (4, 5), (6, 7), (8, 9)] Incorrect usage: >>> numbers = db.range(31, npartitions=1) >>> fizz = numbers.filter(lambda n: n % 3 == 0) >>> buzz = numbers.filter(lambda n: n % 5 == 0) >>> fizzbuzz = db.zip(fizz, buzz) >>> list(fizzbuzz) [(0, 0), (3, 5), (6, 10), (9, 15), (12, 20), (15, 25), (18, 30)] When what you really wanted was more along the lines of the following: >>> list(fizzbuzz) # doctest: +SKIP (0, 0), (3, None), (None, 5), (6, None), (9, None), (None, 10), (12, None), (15, 15), (18, None), (None, 20), (21, None), (24, None), (None, 25), (27, None), (30, 30) rc3s|]}|jkVqdSrrrSrrrGrJrXszbag_zip..zzip-c s2i|]*fttftfddDfqS)c3s|]}|jfVqdSrrrrrGrJrXsz%bag_zip...)r\rr]rS)rrrrJrgszbag_zip..r)rrLrr*rr4rrrrG)rrrrJbag_zips&   rcCs,|rt|f|}dd|D}t|||dS)aMap ``f`` across one or more iterables, maybe with keyword arguments. Low-level function used in ``bag_map``, not user facing. Arguments --------- f : callable iters : List[Iterable] iter_kwarg_keys : List[str] or None Keyword names to use for pair with the tail end of ``iters``, allowing keyword arguments to be passed in from iterators. kwargs : dict or None Additional constant keyword arguments to use on every call to ``f``. cSsg|] }t|qSrG)rnrSr9rGrGrJrUszmap_chunk..) kwarg_keys)r _MapChunk)r|itersiter_kwarg_keysrorGrGrJ map_chunks rc@s&eZdZdddZddZddZdS) rNcCs&||_||_|pd|_t|j|_dSr)r|rrr[nkws)rIr|rrrGrGrJrs z_MapChunk.__init__cCs|zdd|jD}Wntk r2|YnX|jrr|d|j }tt|j||j d}|j||S|j|S)NcSsg|] }t|qSrG)rrrGrGrJrUsz&_MapChunk.__next__..)r StopIterationcheck_all_iterators_consumedrrrrr|)rIvalsrrorGrGrJ__next__s z_MapChunk.__next__c CsZt|jdkrV|jD]@}t|tjr&qz t|Wntk rFYqXd}t|qdS)NrVzmap called with multiple bags that aren't identically partitioned. Please ensure that all bag arguments have the same partition lengths)r[rrvrarepeatrrr)rIrr`rGrGrJrs   z&_MapChunk.check_all_iterators_consumed)N)rrrrrrrGrGrGrJrs  rcCs|rt|f|}t||Sr)rrar)r|rOrorGrGrJrs rcCsli}g}|D]4\}}t|\}}|s2|||<q|||<||q|rdttt|t|ff}||fS)zExtracts dask values from kwargs. Currently only ``dask.bag.Item`` and ``dask.delayed.Delayed`` are supported. Returns a merged dask graph and a task resulting in a keyword dict. )rjr3rrrrZr)roZkwargs2rmrerfvvrrGrGrJrs   rc sdttdf||g}g}g|D]`}t|trR|||q.t|ttfr~||tj |j fq.tj |fq.ii| D].\}}t|tr||<||q||<qt \}| ||stddd|D} t| dkrtd| } fdd r>> import dask.bag as db >>> b = db.from_sequence(range(5), npartitions=2) >>> b2 = db.from_sequence(range(5, 10), npartitions=2) Apply a function to all elements in a bag: >>> db.map(lambda x: x + 1, b).compute() [1, 2, 3, 4, 5] Apply a function with arguments from multiple bags: >>> from operator import add >>> db.map(add, b, b2).compute() [5, 7, 9, 11, 13] Non-bag arguments are broadcast across all calls to the mapped function: >>> db.map(add, b, 1).compute() [1, 2, 3, 4, 5] Keyword arguments are also supported, and have the same semantics as regular arguments: >>> def myadd(x, y=0): ... return x + y >>> db.map(myadd, b, y=b2).compute() [5, 7, 9, 11, 13] >>> db.map(myadd, b, y=1).compute() [1, 2, 3, 4, 5] Both arguments and keyword arguments can also be instances of ``dask.bag.Item`` or ``dask.delayed.Delayed``. Here we'll add the max value in the bag to each element: >>> db.map(myadd, b, b.max()).compute() [4, 5, 6, 7, 8] rr$At least one argument must be a Bag.cSsh|] }|jqSrGrrSrrGrGrJriszbag_map..rV1All bags must have the same number of partitions.cs6fddD}r2|fddD|S)Ncs$g|]}t|tr|jfn|qSrGrvrrrrrGrJrUsz0bag_map..build_iters..c3s|]}|jfVqdSrrrrrGrJrXsz/bag_map..build_iters..)rr)rr)args2 bag_kwargsrrJ build_itersszbag_map..build_itersNc s(i|] }|ftt|ffqSrG)r\rrSr)rrrr other_kwargsrGrJrgs zbag_map..r)rr>r*rvrrrr2rarrrjrrrr[r@rZrrrrYr4r) rrrormrr9rerfrrrb return_typerrG)rrrrrrrrJr)sL8           rcsL|ddptdtdf||g}gg}|D]T}t|trb|||q>t|ttfr|j ||q>|q>ii| D].\}}t|tr||<||q||<qt \}| ||st ddd|D} t| dkrt d | } fd d fd d r`fddt| D} ng} i} D]@} t| tr| | jdg| jf| | j<n| | dgqlrttrddddddtdf| | d|d} ttt|}t|dkr(|nt}tj| ||d}||| S)aApply a function to every partition across one or more bags. Note that all ``Bag`` arguments must be partitioned identically. Parameters ---------- func : callable *args, **kwargs : Bag, Item, Delayed, or object Arguments and keyword arguments to pass to ``func``. Examples -------- >>> import dask.bag as db >>> b = db.from_sequence(range(1, 101), npartitions=10) >>> def div(nums, den=1): ... return [num / den for num in nums] Using a python object: >>> hi = b.max().compute() >>> hi 100 >>> b.map_partitions(div, den=hi).take(5) (0.01, 0.02, 0.03, 0.04, 0.05) Using an ``Item``: >>> b.map_partitions(div, den=b.max()).take(5) (0.01, 0.02, 0.03, 0.04, 0.05) Note that while both versions give the same output, the second forms a single graph, and then computes everything at once, and in some cases may be more efficient. rANrzmap-partitionsrcSsh|] }|jqSrGrrrGrGrJrisz!map_partitions..rVrcsfddDS)Ncs$g|]}t|tr|jfn|qSrGrrrrGrJrUsz6map_partitions..build_args..rGr)rrrJ build_argssz"map_partitions..build_argscs,siStttfddDffS)Ncsg|]}|jfqSrGrrrrGrJrUsz.build_bag_kwargs..)rrrZrr)rrrJbuild_bag_kwargss z(map_partitions..build_bag_kwargsc s,i|]$}|ft|t|ffqSrG)r9rr)rrrrrrGrJrgs z"map_partitions..rrT) numblocksZ concatenatermr)r@r>rr*rvrrrr2rrjrrrr[rrrr]r+rrrYr4r)rrrorrmr9rerfrrrbpairsrrTrrrG)rrrrrrrrJrsx#               $ rcCs"|tk rt|||St||SdSr)rr )r)rr$rGrGrJr(" s r(csfdd}|S)Ncs|dSrrG)rOrestagerGrJh* szmake_group..hrG)rerrrGrrJ make_group) sr c s&|pd}|j}ttt|t|p.ddkrPtt|dn|g}g}g}fddtD}|ttj |d} t |||} d| d| } d| i} t |D]\} }i}i}| |jkr| j | f| d |f<ng| d |f<tddD]\| |f}t tdfd|ff||<tD] }tj||if|||f<qHq||||qtddD](fd d |D}||qd | fd d t |D}||||t| |f|}tj|| gd}t||t|S)NrrVcs(g|] tfddtDqS)c3s|]}t|VqdSr)r:r<)rrerGrJrX> sz+groupby_tasks...rr)restagesrrJrU> sz!groupby_tasks..)rrwz shuffle-join-zshuffle-group-zshuffle-split-rcs8i|]0fttjfddtDffqS)c s,g|]$}dtd|fqSrV)r?r<)inpshuffle_split_namerrGrJrUk s  z,groupby_tasks...)rZrrrr)reshuffle_join_namerr)rrJrgf sz!groupby_tasks..zshuffle-c s0i|](\}}|fttjttd|ffffqSr)rZrrjrrrc)rwrrGrJrg} sr)rrrXrlogrrrr#Zgroupby_tasks_group_hashr*rrrrrrrrrr4rrYr[)rrwrrsrgroupssplitsZjoinsinputsZb2rAZshuffle_group_namerWidxrrsplitZ _key_tuplerrendrbrrG)rwrerrrrrrJrv0 sd"          rvcs<dkrjt}ddl}d|ftdd}|rRt|jdd|if}n|jf}z|j|j|ffi}Wn"t k r|j|fi}YnXdt d|fdd t jD} d |t j ft| i} d |fd d t D} t|| | | } tj| gd } t| S)Nrzpartd-Ztemporary_directoryrGrz groupby-part-r c s(i|] }|ftj|ffqSrG)rrr)rrrrwrrrrGrJrg sz groupby_disk..zgroupby-barrier-zgroupby-collect-cs i|]}|ft|fqSrG)rr)rrwrrrGrJrg sr)rr*partdr"rr9ZFilePythonZSnappyrr>rr#Zbarrierr]rr4rrY)rrwrrrrArdirnamefileZdsk1rprqrrrbrrG)rrrrrwrrrrJru s4  rucCsdt|trDzt|\}}Wn tk r:|s6tYSYnX||S|sXt|dkrXtS||SdSr)rvrrrrFr[)rrr:rrGrGrJr8 s r8cCsdd|D}t|||S)Ncss|]}|tk r|VqdSr)rF)rSrrGrGrJrX sz'empty_safe_aggregate..)r8)rrr:parts2rGrGrJr? sr?cCs>tt||}t||kr:|r:td|dt|d|S)Nz"Insufficient elements for `take`. z elements requested, only z@ elements available. Try passing larger `npartitions` to `take`.)rZrr[warningsrl)rrrlrrGrGrJrk s rkccs0t}|||D]}||kr|VqdS)a7Filter elements of `x` by a probability `prob`. Parameters ---------- x : iterable state_data : tuple A tuple that can be passed to ``random.Random.setstate``. prob : float A float between 0 and 1, representing the probability that each element will be yielded. N)r setstaterandom)rOrrrrrGrGrJr s   rcs.ttstdfddt|DS)aReturn a list of tuples that can be passed to ``random.Random.setstate``. Parameters ---------- n : int Number of tuples to return. random_state : int or ``random.Random``, optional If an int, is used to seed a new ``random.Random``. lcs2g|]*}dtfddtdDddfqS)c3s|]}dVqdS)rN)randintrZ maxuint32rrGrJrX sz6random_state_data_python...p)rNrrrrGrJrU s z,random_state_data_python..)rvr r)rrrGrrJr s  rcs`tttfstt|fddt|dD}|t|dd|S)zvSplit apart a sequence into n equal pieces. >>> split(range(10), 3) [[0, 1, 2], [3, 4, 5], [6, 7, 8, 9]] cs,g|]$}t|t|dqSrrrrrMrGrJrU szsplit..rVN)rvrZr]r[rrr)rMrLrGrrJr s  rcCsBddl}t|}t|ts"t|}|j|t|d}|j|ddS)NrryF)copy)r{r\rvrZr|Zastype)rMrzr~rrrGrGrJr s  rcs||jkr|Sd|t||f}|j|krZ|j|fddt|dD}t|||St||j\}}|g|j}|d|7<t|||SdS)zChanges the number of partitions of the bag. This can be used to reduce or increase the number of partitions of the bag. zrepartition-%d-%scsg|]}t|qSrGrrSZold_partition_indexZratiorGrJrU" sz+repartition_npartitions..rVriN)rr*r_repartition_from_boundariesdivmod_split_partitions)rrnew_namenew_partitions_boundariesdivmodnsplitsrGrrJr s       rcCs6ddlm}ddlm}t||r.t||}t|S)Nr)deepcopy)chain)rrrarrvr\r8)rrrrGrGrJtotal_mem_usage. s     rc sttrtt|t}fdd|D}tdd|Drdt|}t |||}g}t ||D]\}}| ||g|qv|}t fdd|Dst tttt|}ttj|} dt|} t|| | S)z\ Repartition bag so that new partitions have approximately `size` memory usage each csg|]}d|qSrrGrSZ mem_usagerrGrJrUC sz$repartition_size..css|]}|dkVqdS)rVNrG)rSZnsplitrGrGrJrXD sz#repartition_size..zrepartition-split-c3s|]}|kVqdSrrGrrrGrJrXN sz repartition-)rvrrBrrrrrKr*rrrrLrrZrr[r@rraddr) rrZ mem_usagesr split_nameZsplit_mem_usagesrusageZnew_npartitionsrrrGrrJr9 s"   rc Cst||jkrtd|ji}dt||}d}t|D]r\}}|dkrj|j|f|||f<|d7}q>t|j|f|f|||f<t|D]$}tj ||f|f|||f<|d7}qq>t j |||gd} t | |t |dS)aDSplit a Dask bag into new partitions Parameters ---------- bag: Dask bag nsplits: List[int] Number of target bags for each partition The length of nsplits should be the same as bag.npartitions new_name: str See Also -------- repartition_npartitions repartition_size znsplits should have len=zsplit-rrVrrr)r[rrr*rrrrrrr4rrrG) rrrrbrr=rreZjjrrGrGrJrU s   rc st|tst|}|ddkr*|dd|djkrD|jt|d}i}t|D]>}ttjfddt||||dDff}||||f<q\t j ||gd}t |||dS)NrrirVcsg|]}j|fqSrGrrrrGrJrU sz0_repartition_from_boundaries..rr) rvrZr?rrr[rrrr4rr)rrrZnum_new_partitionsrbZnew_partition_indexrrrGrrJrx s,         r)T)N)NN)F)rp)NN)N)N)NN)Nrp)T)N) __future__rrwrarXrrrrrcollections.abcrrrr functoolsrr r rr urllib.requestr ZtlzrZ fsspec.corer rrrrrrrrrrrrrrrrrrr r!rr"Zdask.bagr#Z dask.bag.avror%Z dask.baser&r'r(r)r*Zdask.blockwiser+Z dask.contextr,Z dask.corer-r.r/r0r1rr2r3Zdask.highlevelgraphr4Zdask.optimizationr5r6r7Z dask.sizeofr8Z dask.utilsr9r:r;r<r=r>r?r@rArBrCrDrrrrYrrFrQrcrnrsrrrrrrrrrrrrrr/rr\rr4r5r.rrrrrrrrr(rrrvrur8r?rkrrrrrrrrrrGrGrGrJs     X       8     ] 9 X  /  +  3 &w Z )    #