o bic@sddlZddlZddlZddlmZmZddlmZddlm Z m Z m Z m Z m Z mZmZmZmZmZmZddlZddlZddlZddlmZddlmZmZddlmZddl m!Z!dd l"m#Z#e rddl$Z$ddlZdd l%m&Z&dd l'm(Z(dd l)m*Z*dd l+m,Z,edddZ-edddZ.edZ/edZ0edZ1ee2ddfZ3edZ4e5e6Z7e#GdddeZ8edde e9ej:ffZ;ee4ej:fZee e-ge.fe e-ge e.fdfZ?eeee1dfZ@edZAgd ZBd!ZCd"ee9d#e9fd$d%ZDe#d&edd#ed'fd(d)ZEe#Gd*d+d+ZFGd,d-d-ZGe#eGd.d'd'ZHd/d0eeHDZIe#eGd1ddeHZJe#d2d3eGd4d5d5eJZKe#Gd6d7d7ZLe#d8d3Gd9d:d:ZMd;eNej:d#ej:fdN) dataclassfields)Enum) TYPE_CHECKINGAnyCallableDictIteratorListOptionalProtocolTupleTypeVarUnion)ArrowConversionError)_check_pyarrow_version_truncated_repr) ObjectRef)log_once) DeveloperAPI) BlockBuilder)PandasBlockSchema)SortKey) AggregateFnTT) contravariantU) covariantKeyTypeAggType) pyarrow.Tablepandas.DataFramerpyarrow.lib.Schema)zpyarrow.ChunkedArray pyarrow.Arrayz pandas.Seriesc@seZdZdZdZdS) BlockTypearrowpandasN)__name__ __module__ __qualname__ARROWPANDASr,r,B/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/block.pyr$@sr$r r!c@s*eZdZdedeeeeffddZdS)_CallableClassProtocol_CallableClassProtocol__argreturncCsdSNr,)selfr/r,r,r-__call__Tsz_CallableClassProtocol.__call__N)r'r(r)rrrr r3r,r,r,r-r.Ss"r. BlockMetadata)r&pyarrownumpyNr6given_batch_formatr0cCs.|dkrt}|tvrtd|dtd|S)NdefaultzThe given batch format z isn't allowed (must be one of z).)DEFAULT_BATCH_FORMATVALID_BATCH_FORMATS ValueError)r7r,r,r-_apply_batch_formatksr<metas BlockStatscCsdd|DS)NcSsg|]}|qSr,)to_stats).0mr,r,r- xszto_stats..r,)r=r,r,r-r?vsr?c@s.eZdZdZddZed ddZdd Zd S) BlockExecStatsaExecution stats for this block. Attributes: wall_time_s: The wall-clock time it took to compute this block. cpu_time_s: The CPU time it took to compute this block. node_id: A unique id for the node that computed this block. max_uss_bytes: An estimate of the maximum amount of physical memory that the process was using while computing this block. cCs>d|_d|_d|_d|_d|_tj|_ d|_ d|_ dS)Nr) start_time_s end_time_s wall_time_s udf_time_s cpu_time_srayruntime_contextget_runtime_context get_node_idnode_id max_uss_bytestask_idxr2r,r,r-__init__s zBlockExecStats.__init__r0_BlockExecStatsBuildercCtSr1)rRr,r,r,r-builderzBlockExecStats.buildercCst|j|j|j|jdS)N)rFrHrGrM)reprrFrHrGrMrPr,r,r-__repr__szBlockExecStats.__repr__N)r0rR)r'r(r)__doc__rQ staticmethodrTrWr,r,r,r-rC{s    rCc@s"eZdZdZddZd ddZdS) rRzHelper class for building block stats. When this class is created, we record the start time. When build() is called, the time delta is saved as part of the stats. cCst|_t|_dSr1)time perf_counter _start_time process_time _start_cpurPr,r,r-rQs z_BlockExecStatsBuilder.__init__r0rCcCs@t}t}t}|j|_||_||j|_||j|_ |Sr1) rZr[r]rCr\rDrErFr^rH)r2end_timeend_cpustatsr,r,r-builds  z_BlockExecStatsBuilder.buildN)r0rC)r'r(r)rXrQrbr,r,r,r-rRsrRc@s>eZdZUdZeeed<eeed<eeed<ddZdS)r>z#Statistics about the block producednum_rows size_bytes exec_statscCs"|jdur t|jtsJdSdSr1)rd isinstanceintrPr,r,r- __post_init__s zBlockStats.__post_init__N) r'r(r)rXr rg__annotations__rCrhr,r,r,r-r>s     cCsh|]}|jqSr,)name)r@fr,r,r- srlcs:eZdZUdZeeeed<ddZfddZ Z S)r4zMetadata about the block. input_filescstdifddtDS)Ncsi|]}||qSr,)__getattribute__)r@keyrPr,r- sz*BlockMetadata.to_stats..r,)r>_BLOCK_STATS_FIELD_NAMESrPr,rPr-r?szBlockMetadata.to_statscs"t|jdurg|_dSdSr1)superrhrmrP __class__r,r-rhs   zBlockMetadata.__post_init__) r'r(r)rXr r strrir?rh __classcell__r,r,rsr-r4s alpha) stabilitycsneZdZUdZeeed<ddededffdd Z dde ded d dfd d Z e d efd dZ Z S)BlockMetadataWithSchemaNschemametadataSchemacs&tj|j|j|j|jd||_dS)N)rmrdrcre)rrrQrmrdrcrerz)r2r{rzrsr,r-rQs z BlockMetadataWithSchema.__init__blockrarCr0cCs*t|}|j|d}|}t||dS)N)re)r{rz) BlockAccessor for_block get_metadatarzry)r}raaccessormetarzr,r,r- from_blocks   z"BlockMetadataWithSchema.from_blockcCst|j|j|j|jdS)N)rcrdrerm)r4rcrdrermrPr,r,r-r{s z BlockMetadataWithSchema.metadatar1)r'r(r)rzr r|rir4rQBlockrpropertyr{rvr,r,rsr-rys   ryc@seZdZdZdefddZdedeefddZ drd ed ed ede fd d Z de ede fddZ de eede fddZdeeefde fddZdeede fddZdsddZ dtdeeee efdeejeeejfffddZdud!d"Zde fd#d$Zde fd%d&Zd'eedefd(d)Zdefd*d+Zdeed,ffd-d.Z  dvd/ee ed0ee!de"fd1d2Z#dwd5d6Z$e%dxd8d9Z&e' dtd:ed;ee(de fdd?Z+e'd:eee*fde fd@dAZ,e%dBe ddCfdDdEZ-dFedGdHdd4fdIdJZ.drdKedLedee/fdMdNZ0dKedLedee/fdOdPZ1dKedLedee/fdQdRZ2dKedLedee/fdSdTZ3dKedLedee/fdUdVZ4 dtdKedLedWee/dee/fdXdYZ5dydZd[Z6d\e edGdHde d4fd]d^Z7d_dHd`e8dade fdbdcZ9e%dde d4dGdHde8e e:ffdedfZ;e% gdzdde e dGdHd`e8dadhede8e e:ff didjZdoe edejfdpdqZ?dS){r~aProvides accessor methods for a specific block. Ideally, we wouldn't need a separate accessor classes for blocks. However, this is needed if we want to support storing ``pyarrow.Table`` directly as a top-level Ray object, without a wrapping class (issue #17186). r0cCt)z2Return the number of rows contained in this block.NotImplementedErrorrPr,r,r-rczBlockAccessor.num_rowspublic_row_formatcCr)zIterate over the rows of this block. Args: public_row_format: Whether to cast rows into the public Dict row format (this incurs extra copy conversions). r)r2rr,r,r- iter_rowszBlockAccessor.iter_rowsFstartendcopycCr)a(Return a slice of this block. Args: start: The starting index of the slice (inclusive). end: The ending index of the slice (exclusive). copy: Whether to perform a data copy for the slice. Returns: The sliced block result. r)r2rrrr,r,r-slices zBlockAccessor.sliceindicescCr)zReturn a new block containing the provided row indices. Args: indices: The row indices to return. Returns: A new block containing the provided row indices. r)r2rr,r,r-take)s zBlockAccessor.takecolumnscCr)z3Return a new block containing the provided columns.rr2rr,r,r-select4rzBlockAccessor.selectcolumns_renamecCr)z0Return the block reflecting the renamed columns.r)r2rr,r,r-rename_columns8rzBlockAccessor.rename_columns random_seedcCr)zRandomly shuffle this block.r)r2rr,r,r-random_shuffle<rzBlockAccessor.random_shuffler!cCr)z+Convert this block into a Pandas dataframe.rrPr,r,r- to_pandas@rzBlockAccessor.to_pandasNcCr)zConvert this block (or columns of block) into a NumPy ndarray. Args: columns: Name of columns to convert, or None if converting all columns. rrr,r,r-to_numpyDzBlockAccessor.to_numpyr cCr)z'Convert this block into an Arrow table.rrPr,r,r-to_arrowNrzBlockAccessor.to_arrowcCr)z/Return the base block that this accessor wraps.rrPr,r,r-to_blockRrzBlockAccessor.to_blockcCs|S)z1Return the default data format for this accessor.)rrPr,r,r- to_defaultVszBlockAccessor.to_default batch_formatcCsl|dur|S|dks|dkr|S|dkr|S|dkr$|S|dkr,|Stdtd|) zConvert this block into the provided batch format. Args: batch_format: The batch format to convert this block to. Returns: This block formatted as the provided batch format. Nr8nativer&r5r6z The batch format must be one of z, got: )rrrrrr;r:)r2rr,r,r-to_batch_formatZs zBlockAccessor.to_batch_formatcCr)z3Return the approximate size in bytes of this block.rrPr,r,r-rdsrzBlockAccessor.size_bytesr"cCr)z7Return the Python type or pyarrow schema of this block.rrPr,r,r-rzwrzBlockAccessor.schemarmrecCst||||dS)z)Create a metadata object from this block.)rcrdrmre)r4rcrd)r2rmrer,r,r-r{s zBlockAccessor.get_metadataotherrcCr)z array, e.g., `{'data': array}` instead of `array`.N!_fallback_to_pandas_block_warningz)Failed to convert batch to Arrow due to: z; falling back to Pandas block)rfnpndarrayr;r collectionsabcMappingr$r*batch_to_arrow_blockrrloggerwarningbatch_to_pandas_blockr+)clsrrer,r,r-batch_to_blocks*    zBlockAccessor.batch_to_blockcCddlm}||S)z4Create an Arrow block from user-facing data formats.r)ArrowBlockBuilder)ray.data._internal.arrow_blockr_table_from_pydict)rrrr,r,r-r  z"BlockAccessor.batch_to_arrow_blockcCr)z4Create a Pandas block from user-facing data formats.r)PandasBlockBuilder)ray.data._internal.pandas_blockrr)rrrr,r,r-rrz#BlockAccessor.batch_to_pandas_blockr}zBlockAccessor[T]cCstddl}ddl}t||jrddlm}||St||jr+ddlm }||St|t r;ddlm}| |St|t rJt dt|dtd|t|)z,Create a block accessor for the given block.rN)ArrowBlockAccessor)PandasBlockAccessorrz: Standalone Python objects are not allowed in Ray 2.5. To use Python objects in a dataset, wrap them in a dict of numpy arrays, e.g., return `{'item': batch}` instead of just `batch`.zNot a block type: {} ({}))rr&r5rfTablerr DataFramerrbytes from_byteslistr;r TypeErrorformattype)r}r&r5rrr,r,r-rs"        zBlockAccessor.for_block n_samplessort_keyrcCr)z0Return a random sample of items from this block.r)r2rrr,r,r-samplerzBlockAccessor.sampleon ignore_nullscCr)z=Returns a count of the distinct values in the provided columnrr2rrr,r,r-countrzBlockAccessor.countcCr)z2Returns a sum of the values in the provided columnrrr,r,r-sumrzBlockAccessor.sumcCr)z2Returns a min of the values in the provided columnrrr,r,r-minrzBlockAccessor.mincCr)z2Returns a max of the values in the provided columnrrr,r,r-maxrzBlockAccessor.maxcCr)z3Returns a mean of the values in the provided columnrrr,r,r-meanrzBlockAccessor.meanrcCr)zBReturns a sum of diffs (from mean) squared for the provided columnr)r2rrrr,r,r-sum_of_squared_diffs_from_meanrz,BlockAccessor.sum_of_squared_diffs_from_meancCr)z9Returns new block sorted according to provided `sort_key`r)r2rr,r,r-sortrzBlockAccessor.sort boundariescCr)z1Return a list of sorted partitions of this block.rr2rrr,r,r-sort_and_partitionsz BlockAccessor.sort_and_partitionroaggsrcCr)z3Combine rows with the same key into an accumulator.r)r2rorr,r,r- _aggregate rzBlockAccessor._aggregateblockscCr)z9Return a sorted block by merging a list of sorted blocks.r)rrr,r,r-merge_sorted_blockssz!BlockAccessor.merge_sorted_blocksTfinalizecCr)z/Aggregate partially combined and sorted blocks.r)rrrrr,r,r-_combine_aggregated_blocksrz(BlockAccessor._combine_aggregated_blockscCr)zNOTE: PLEASE READ CAREFULLY Returns dataset partitioned using list of boundaries This method requires that - Block being sorted (according to `sort_key`) - Boundaries is a sorted list of tuples rrr,r,r-_find_partitions_sorted!s z%BlockAccessor._find_partitions_sortedcCr)z$Return the block type of this block.rrPr,r,r-r0rzBlockAccessor.block_typekeyscCsL|dkrtjgtjdS|std|gS||}tt|S)a NOTE: THIS METHOD ASSUMES THAT PROVIDED BLOCK IS ALREADY SORTED Compute boundaries of the groups within a block based on provided key (a column or a list of columns) NOTE: In each column, NaNs/None are considered to be the same group. Args: block: sorted block for which grouping of rows will be determined based on provided key keys: list of columns determining the key for every row based on which the block will be grouped Returns: A list of starting indices of each group and an end index of the last group, i.e., there are ``num_groups + 1`` entries and the first and last entries are 0 and ``len(array)`` respectively. r)dtype)rcrarrayint32r"_get_group_boundaries_sorted_numpyrvalues)r2rprojected_blockr,r,r-_get_group_boundaries_sorted4s  z*BlockAccessor._get_group_boundaries_sortedF)r0r!r1)r0r )NN)rrr0r)r0r)rrr0r)T)@r'r(r)rXrgrcboolr rrrrr rr rurrrrrrrrrrrr DataBatchrrdrrzrCr4rrrYrT classmethodr$rrrrrrrrrrrrrrrr rryrrrrrr,r,r,r-r~s         #        r~betac @seZdZdZdefddZdddeded eefd d Z dddeded eefd d Z dddeded eefddZ dddeded eefddZ dddeded eefddZ dddededed eefddZd efddZd efddZddddedeeded eefddZd eefd d!Zd+d#ed ejfd$d%Zd eeed&ffd'd(Zeded dfd)d*ZdS),BlockColumnAccessorzbProvides vendor-neutral interface to apply common operations to block's (Pandas/Arrow) columnscolcCs ||_dSr1)_column)r2rr,r,r-rQ[s zBlockColumnAccessor.__init__T)as_pyrrr0cCt)z4Returns a count of the distinct values in the columnrr2rrr,r,r-r^rUzBlockColumnAccessor.countcCrS)z)Returns a sum of the values in the columnrrr,r,r-rbrUzBlockColumnAccessor.sumcCr)z)Returns a min of the values in the columnrrr,r,r-rfrUzBlockColumnAccessor.mincCr)z)Returns a max of the values in the columnrrr,r,r-rjrUzBlockColumnAccessor.maxcCr)z*Returns a mean of the values in the columnrrr,r,r-rnrUzBlockColumnAccessor.meanqcCr)z.Returns requested quantile of the given columnr)r2rrrr,r,r-quantilerszBlockColumnAccessor.quantilecCr)zBReturns new column holding only distinct values of the current onerrPr,r,r-uniquexrUzBlockColumnAccessor.uniquecCr)z;Flattens nested lists merging them into top-level containerrrPr,r,r-flatten|szBlockColumnAccessor.flattenN)rrrcCr)z9Returns a sum of diffs (from mean) squared for the columnr)r2rrrr,r,r-rsz2BlockColumnAccessor.sum_of_squared_diffs_from_meancCr)z8Converts block column to a list of Python native objectsrrPr,r,r- to_pylistrUzBlockColumnAccessor.to_pylistFzero_copy_onlycCr)z#Converts underlying column to Numpyr)r2rr,r,r-rrUzBlockColumnAccessor.to_numpyr#cCr)zAConverts block column into a representation compatible with ArrowrrPr,r,r-_as_arrow_compatiblerUz(BlockColumnAccessor._as_arrow_compatiblecCsntddl}t|tjst|tjrddlm}||St||jr-ddl m }||St dt |d)z-Create a column accessor for the given columnrN)ArrowBlockColumnAccessor)PandasBlockColumnAccessorzEExpected either a pandas.Series or pyarrow.Array (ChunkedArray) (got )) rr&rfpaArray ChunkedArrayrrSeriesrrrr)rpdrrr,r,r- for_columns   zBlockColumnAccessor.for_columnr)r'r(r)rX BlockColumnrQrr rrrrrrfloatrrrrr rrrrrrrrYrr,r,r,r-rVsH         rrcCs@g}g}g}|D]0}t|jtjrt|dr||qt|jtjs3|ddur3||q||qg}t|dkrR|tdd|Djddt|dkri|tdd|Djddt|dkr|tdd|Djddt dgt |jdd ddt|dgg t }|S) NrcSs$g|]}|dd|ddkqSNrr,r@arrr,r,r-rBs$z6_get_group_boundaries_sorted_numpy..)axisc SsHg|] }|dd|ddkt|ddt|ddB@qSr)risfiniterr,r,r-rBs "c SsNg|]#}|dd|ddkt|dddt|ddd@@qSr)requalrr,r,r-rBs (r)r issubdtypernumberisnanappendlenvstackanyhstack column_stacknonzeroastyperg)rgeneral_arraysnum_arrays_with_nancat_arrays_with_nonerdiffsrr,r,r-rsV       r)PrloggingrZ dataclassesrrenumrtypingrrrrr r r r r rrr6rr5rrI$ray.air.util.tensor_extensions.arrowrray.data._internal.utilrr ray.typesrray.utilrray.util.annotationsrr& ray.data._internal.block_builderrrr2ray.data._internal.planner.exchange.sort_task_specrray.data.aggregaterrrrrrrr|r getLoggerr'rr$rurrDataBatchColumn CallableClassr.UserDefinedFunctionBlockPartitionBlockPartitionMetadatar:r9r<r?rCrRr>rqr4ryr~rrrr,r,r,r-s 4             $QV