o Nrf#@sLddlZddlZddlZddlZddlmZ GdddZ GdddZ dS)Nc@seZdZddZddZddZddZd d Zd d Zd dZ ddZ ddZ ddZ ddZ ddZddZddZddZdd Zd!S)" ParallelDatacCs ||_dSNdata)selfrrV/var/www/html/software/conda/envs/catlas/lib/python3.10/site-packages/clodius/fpark.py__init__  zParallelData.__init__cstfdd|jDS)Ncsg|]}|qSrr).0dfuncrr sz$ParallelData.map..)rrrrrr rmap szParallelData.mapcCs t|jSr)lenrrrrrcountr zParallelData.countcCs|j|j|_|Srr)rZ other_datarrrunionszParallelData.unioncCsdS)NrrrrrgetNumPartitionszParallelData.getNumPartitionscCs.tt}|jD] }||dd7<q|S)zE Count the number of elements with a particular key. rr)col defaultdictintr)rcountsr rrr countByKeys  zParallelData.countByKeycCs|Srr)rnumrrrcoalesce$rzParallelData.coalescecCs8tt}|jD] }||d|dqt|S)zh When called on a dataset of (K, V) pairs, returns a dataset of (K, Iterable) pairs. rr)rrlistrappendritems)rbucketsr rrr groupByKey's   zParallelData.groupByKeycCs t||j}tttj|S)zd Flatten a list of results mapped to the data and return as one large list. )rrrr itchain from_iterable)rrresultrrrflatMap2s zParallelData.flatMapcCs|jD]}||qdSrr)rrr rrrforeach;s  zParallelData.foreachcCs||jdSrrrrrrforeachPartition?zParallelData.foreachPartitioncCs\tt}|jD] }||d|dqt}|D] }t|||||<qt| S)aV When called on a dataset of (K, V) pairs, returns a dataset of (K, V) pairs where the values for each key are aggregated using the given reduce function func, which must be of type (V,V) => V. Like in groupByKey, the number of reduce tasks is configurable through an optional second argument. rr) rrr rr!dictftreducerr")rrr#r reduced_bucketskeyrrr reduceByKeyBs   zParallelData.reduceByKeyc Csptt}|jD] }||d|dqt}|D]}|}||D]} ||| }q%|||<qt|S)Nrr) rrr rr!r-copyrr") rZ start_valZseq_funcZ comb_funcr#r r0r1Zcomb_valvalrrraggregateByKeyTs      zParallelData.aggregateByKeycCst||jSr)r.r/rrrrrr/br,zParallelData.reducecCs|jSrrrrrrcollecteszParallelData.collectcCs|jd|Srr)rnrrrtakehr,zParallelData.takeN)__name__ __module__ __qualname__r rrrrrrr$r)r*r+r2r5r/r6r8rrrrrs"    rc@s<eZdZdZddZeddZeddZedd Zd S) FakeSparkContextz6 Emulate a SparkContext for local processing. cCsdSrrrrrrr qrzFakeSparkContext.__init__cCst|Sr)rrrrr parallelizetszFakeSparkContext.parallelizecCsHt|d}tttdd|WdS1swYdS)z Load a single file as a ParallelData set. :param filename: A file to be loaded line by line @return: A ParallelData object wrapping the lines in the file. rcSs|Sr)strip)xrrrsz1FakeSparkContext.singleTextFile..N)openrr r readlines)filenamefrrrsingleTextFilexs $zFakeSparkContext.singleTextFilecCsNt|r"tt|d}tg}|D] }|t|}q|St|S)a{ Load a filename as a text file. Filename can be either a single file or a directory containing a multitude of 'part-*' files. :param filename: The name of the file (or directory) containing the data which we want to load one by one :return: A ParallelData object containing all of the lines of the file or files zpart-*)opisdirglobjoinr<r=rrF)rDZ parts_filesprrrtextFiles   zFakeSparkContext.textFileN) r9r:r;__doc__r staticmethodr=rFrLrrrrr<ls   r<) collectionsr functoolsr.rI itertoolsr%Zos.pathpathrGrr<rrrrs d