biddlmZddlZddlZddlZddlZddlmZddl m Z ddl m Z m Z ejeZGddZddZd Zd Zdd Zd Zd ZdZej2dZdZdZy)) annotationsN)config) PANDAS_GE_300)group_split_dispatchhash_object_dispatchc$eZdZdZddZdZdZy)maybe_buffered_partdz[ If serialized, will return non-buffered partd. Otherwise returns a buffered partd Nc|xstjdd|_||_tjdd|_||_|ddl}|j|_yy)Ntemporary_directoryzdataframe.shuffle.compressionr)rgettempdirbuffer compression encode_clspartd PandasBlocks)selfrrr rs Q/home/cdr/jupyterlab/.venv/lib/python3.12/site-packages/dask/dataframe/shuffle.py__init__zmaybe_buffered_partd.__init__sWI&**-BD"I  !::&EtL$   #00DO c|jrt|jd|jffSt|jdffS)NF)r r r)rs r __reduce__zmaybe_buffered_partd.__reduce__s7 <<(4??E4<<*PQ Q(4??E*BC Crc.ddl}tjd|j} |jr t |j |jnd}|j|}|jjj||r||}|jr/|j|j!|j#|S|j|S#t$r*}tdj|j|d}~wwxYw)Nrz.partd)suffixdirzxNot able to import and load {} as compression algorithm.Please check if the library is installed and supported by Partd.)rtempfilemkdtempr rgetattr compressedAttributeError ImportErrorformatFilefile cleanup_filesappendrrBufferDict)rargskwargsrpathpartd_compressioner$s r__call__zmaybe_buffered_partd.__call__%sxT\\B ##(($*:*:; zz$   ''- $T*D ;;??5<< d#CD D??4( ( SSYSY$$T    s.C!! D*%DD)NTN)__name__ __module__ __qualname____doc__rrr.rrr r s1D )rr c||j|d}t|dt|z}|jtj|dz S)a Computes a deterministic index mapping each record to a partition. Identical rows are mapped to the same partition. Parameters ---------- df : DataFrame/Series/Index npartitions : int The number of partitions to group into. cast_dtype : dtype, optional The dtype to cast to to avoid nullability issues Returns ------- partitions : ndarray An array of int64 values mapping each record to a partition. ignore)errorsFindex)astyperintnpmin_scalar_type)df npartitions cast_dtyperess rpartitioning_indexrBGsW&YYz(Y 3 r /#k2B BC ::b((;?);< ==rct|y)Nr)list)r)s rbarrierrEds J rct|5|j|}t|dkDr|n|cdddS#1swYyxYw)z/Collect partitions from partd, yield dataframesrN)ensure_cleanup_on_exceptionr len)ppartmeta barrier_tokenrAs rcollectrMis; $Q '-eeDk#hls---s #9Acz |r|j|ddz }n"t||j|dz dz }|rt|dz nd||dk|t|dz k\z<|j}t|d|}|d k(rt|dz nd||<|S#ttf$rt j t|d}|j }||j }|r|j||ddz ||<n(t||j||dz dz ||<YwxYw) Nright)sider9int32)dtypervalueslast) searchsortedrH TypeError ValueErrorr<emptynotnaisnar)s divisions ascending na_position partitionsnot_nulldivisions_notnanass rset_partitions_prerdps_ "///@1DJY)*@*@*@*QQTUUJ"(IQ Q:Y!1C#CDE &&(C #x %C,76,Ac)nq(qJsO - z "XXc!fG4 779#IOO$56 ,,Qx[w,G!K x I!..q{.IJ x s;BBD:9D:ct|si|fSt|tr|g}|r.|ddk(r&||djtj }n=t |r||n|dt|zjtj }|jdz}t||||}||jddfS)Nr _partitionsFr7r9 ignore_index) rH isinstancestrr:r<rQrr;maxriloc)r>colsrhnpartsindnresult2s rshuffle_group_2rrs r72v $v Q=(ak  * !TDr G#f+ U &    A A"2sALIG BGGBQK rc"|\}}||vr||S|SNr3)g_headigheads rshuffle_group_getrys GAtAvt  rcDt|tr|g}|r|ddk(r ||d}n)t|r||n|d}|r||k7r|t|z}t j |dz}t rinddi} ||zj|fi| ||zz|z}t||||S)aSplits dataframe into groups The group is determined by their final partition, and which stage we are in in the shuffle Parameters ---------- df: DataFrame cols: str or list Column name(s) on which to split the dataframe. If ``cols`` is not "_partitions", hashing will be used to determine target partition stage: int We shuffle dataframes with many partitions we in a few stages to avoid a quadratic number of tasks. This number corresponds to which stage we're in, starting from zero up to some small integer k: int Desired number of splits from this dataframe npartition: int Total number of output partitions for the full dataframe nfinal: int Total number of output partitions after repartitioning Returns ------- out: Dict[int, DataFrame] A dictionary mapping integers in {0..k} to dataframes such that the hash values of ``df[col]`` are well partitioned. rrfFr7rScopyrg) rirjrr;r<r=rr:r) r>rmstagekr?rhnfinalrotypr*s r shuffle_grouprs:$v Q=(ak"t2d85I f +F #C  [1_ -C!RvuoF $3  $ $S 3F 3q%x ?! CC C FFrc#K dy#t$r7 |j#t$rtjdYwxYwwxYww)zEnsure a partd.File is cleaned up. We have several tasks referring to a `partd.File` instance. We want to ensure that the file is cleaned up if and only if there's an exception in the tasks using the `partd.File`. Nz1ignoring exception in ensure_cleanup_on_exception) Exceptiondroplogger exception)rIs rrGrGsU   R FFH  R   P Q  R s9A  A  A $A AA AA  A cD||jvr|j|S|Srt)r8rr>r8s r drop_overlaprs "bhh.2775>6B6rc^||jvr|j|gS|jSrt)r8loc _constructorrs r get_overlaprs)#rxx/2665'?FR__5FFrrt)TrU) __future__r contextlibloggingrnumpyr<daskrdask.dataframe._compatrdask.dataframe.dispatchrr getLoggerr/rr rBrErMrdrrryrcontextmanagerrGrrr3rrrs"0N   8 $/)/)n>: -> &,G^ &7Gr