o biD"@sddlmZmZmZmZddlZddlmZddl m Z m Z m Z ddl mZmZmZeGdddZeGdd d ZGd d d eZeGd d d egee fZeGdddeZdS))CallableIterableListOptionalN)_check_pyarrow_version)Block BlockMetadataSchema) Deprecated DeveloperAPI PublicAPIc@seZdZdZedddZedededfdd Zde fd d Z de efd d Z dededfddZ edefddZedefddZdS) DatasourcezInterface for defining a custom :class:`~ray.data.Dataset` datasource. To read a datasource into a dataset, use :meth:`~ray.data.read_datasource`. returnReadercKst|fi|Sz Deprecated: Implement :meth:`~ray.data.Datasource.get_read_tasks` and :meth:`~ray.data.Datasource.estimate_inmemory_data_size` instead. )_LegacyDatasourceReader)self read_argsrR/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/datasource/datasource.py create_readerszDatasource.create_reader parallelismReadTaskcKtrNotImplementedError)rrrrrr prepare_readszDatasource.prepare_readcCs.t|j}d}||r|dt| }|S)ztReturn a human-readable name for this datasource. This will be used as the names of the read tasks. r N)type__name__endswithlen)rnamedatasource_suffixrrrget_name!s  zDatasource.get_namecCrzReturn an estimate of the in-memory data size, or None if unknown. Note that the in-memory data size may be larger than the on-disk data size. rrrrrestimate_inmemory_data_size+z&Datasource.estimate_inmemory_data_sizecCr)aMExecute the read and return read tasks. Args: parallelism: The requested read parallelism. The number of read tasks should equal to this value if possible. Returns: A list of read tasks that can be executed to read blocks from the datasource in parallel. rrrrrrget_read_tasks2s zDatasource.get_read_taskscCs,t|jtju}t|jtju}| p| SN)rr)r r&)rhas_implemented_get_read_tasks+has_implemented_estimate_inmemory_data_sizerrrshould_create_reader?szDatasource.should_create_readercCdS)z:If ``False``, only launch read tasks on the driver's node.Trr%rrrsupports_distributed_readsMsz%Datasource.supports_distributed_readsN)rr)r __module__ __qualname____doc__r rintrrstrr#rr&r)propertyboolr-r/rrrrr s    r c@s8eZdZdZdeefddZdededfddZd S) raA bound read operation for a :class:`~ray.data.Datasource`. This is a stateful class so that reads can be prepared in multiple stages. For example, it is useful for :class:`Datasets ` to know the in-memory size of the read prior to executing it. rcCrr$rr%rrrr&\r'z"Reader.estimate_inmemory_data_sizerrcCr)aExecute the read and return read tasks. Args: parallelism: The requested read parallelism. The number of read tasks should equal to this value if possible. read_args: Additional kwargs to pass to the datasource impl. Returns: A list of read tasks that can be executed to read blocks from the datasource in parallel. rr(rrrr)cs zReader.get_read_tasksN) rr0r1r2rr3r&rr)rrrrrSsrc@sBeZdZdefddZdeefddZdededfd d Z d S) r datasourcecK||_||_dSr*) _datasource _read_args)rr7rrrr__init__s z _LegacyDatasourceReader.__init__rcCsdSr*rr%rrrr&wsz3_LegacyDatasourceReader.estimate_inmemory_data_sizerrcCs|jj|fi|jSr*)r9rr:r(rrrr)zsz&_LegacyDatasourceReader.get_read_tasksN) rr0r1r r;rr3r&rr)rrrrrrsrc@seZdZdZ ddegeefdededfddZ e d efd d Z e d edfd d Z e d egeeffddZ d eefddZdS)raA function used to read blocks from the :class:`~ray.data.Dataset`. Read tasks are generated by :meth:`~ray.data.Datasource.get_read_tasks`, and return a list of ``ray.data.Block`` when called. Initial metadata about the read operation can be retrieved via the ``metadata`` attribute prior to executing the read. Final metadata is returned after the read along with the blocks. Ray will execute read tasks in remote functions to parallelize execution. Note that the number of blocks returned can vary at runtime. For example, if a task is reading a single large file it can return multiple blocks to avoid running out of memory during the read. The initial metadata should reflect all the blocks returned by the read, e.g., if the metadata says ``num_rows=1000``, the read can return a single block of 1000 rows, or multiple blocks with 1000 rows altogether. The final metadata (returned with the actual block) reflects the exact contents of the block itself. Nread_fnmetadataschemar cCs||_||_||_dSr*) _metadata_read_fn_schema)rr=r>r?rrrr;s zReadTask.__init__rcC|jSr*)r@r%rrrr>zReadTask.metadatacCrCr*)rBr%rrrr?rDzReadTask.schemacCrCr*)rAr%rrrr=rDzReadTask.read_fnccs0|}t|dstd||EdHdS)N__iter__zlRead function must return Iterable[Block], got {}. Probably you need to return `[block]` instead of `block`.)rAhasattrDeprecationWarningformat)rresultrrr__call__s zReadTask.__call__r*)rr0r1r2rrrrrr;r5r>r?r=rJrrrrr~s"  rc@sXeZdZdZdedefddZdeefddZd edee fd d Z de fd d Z dS)RandomIntRowDatasourceaAn example datasource that generates rows with random int64 columns. Examples: >>> import ray >>> from ray.data.datasource import RandomIntRowDatasource >>> source = RandomIntRowDatasource() # doctest: +SKIP >>> ray.data.read_datasource( # doctest: +SKIP ... source, n=10, num_columns=2).take() {'c_0': 1717767200176864416, 'c_1': 999657309586757214} {'c_0': 4983608804013926748, 'c_1': 1160140066899844087} n num_columnscCr8r*_n _num_columns)rrLrMrrrr;r<zRandomIntRowDatasource.__init__rcCs|j|jdS)NrNr%rrrr&sz2RandomIntRowDatasource.estimate_inmemory_data_sizerc stddlg}|j}|j}td||}dtdtdtffdd jdd t |Dj }d}||krct |||}t |d ||ddd } | t||ffd d | |d||7}||ks7|S)NrcountrMrcs<jjtjjttjj||ftjdddt|DdS)N)sizedtypecSsg|]}d|qS)c_r.0irrr szMRandomIntRowDatasource.get_read_tasks..make_block..)names) Table from_arraysnprandomrandintiinfoint64maxrangerSrM)pyarrowrr make_blocks z9RandomIntRowDatasource.get_read_tasks..make_blockcSsi|] }d|dgqS)rVrrrWrrr sz9RandomIntRowDatasource.get_read_tasks..rQ)num_rows size_bytes input_files exec_statscs ||gSr*rre)rgrrsz7RandomIntRowDatasource.get_read_tasks..)r?)rrfrOrPrcr3rr\ from_pydictrdr?minrappendr) rr read_tasksrLrM block_sizer?rYrSmetar)rgrfrr)s>  z%RandomIntRowDatasource.get_read_taskscCr.)zReturn a human-readable name for this datasource. This will be used as the names of the read tasks. Note: overrides the base `Datasource` method. RandomIntrr%rrrr#r'zRandomIntRowDatasource.get_nameN) rr0r1r2r3r;rr&rrr)r4r#rrrrrKs  .rK)typingrrrrnumpyr^ray.data._internal.utilrray.data.blockrrr ray.util.annotationsr r r r rrrrKrrrrs H 7