o ci@sddlZddlmZmZmZmZmZddlZddlm Z ddl m Z m Z m Z mZddlmZddlmZmZeddGd d d ZdS) N)DictListLiteralOptionalUnion) ActorHandle) DataIteratorDatasetExecutionOptions NodeIdStr)ExecutionResources) DeveloperAPI PublicAPIstable) stabilityc@seZdZdZ   ddeedeefdee de fdd Z d e d e fd d Z edeeefdedeeedeeedeeeeff ddZede fddZdS) DataConfigzClass responsible for configuring Train dataset preprocessing. For advanced use cases, this class can be subclassed and the `configure()` method overriden for custom data preprocessing. allNTdatasets_to_splitexecution_optionsenable_shard_localitycCsZt|ts |dkr ||_ntdt|jd|d|p t|_||_ d|_ d|_ dS)aConstruct a DataConfig. Args: datasets_to_split: Specifies which datasets should be split among workers. Can be set to "all" or a list of dataset names. Defaults to "all", i.e. split all datasets. execution_options: The execution options to pass to Ray Data. By default, the options will be optimized for data ingest. When overriding this, base your options off of `DataConfig.default_ingest_options()`. enable_shard_locality: If true, when sharding the datasets across Train workers, locality will be considered to minimize cross-node data transfer. This is on by default. rzV`datasets_to_split` should be a 'all' or a list of strings of dataset names. Received z with value .gN) isinstancelist_datasets_to_split TypeErrortype__name__rdefault_ingest_options_execution_options_enable_shard_locality_num_train_cpus_num_train_gpus)selfrrrr#S/home/ubuntu/.local/lib/python3.10/site-packages/ray/train/_internal/data_config.py__init__s  zDataConfig.__init__num_train_cpusnum_train_gpuscCs||_||_dS)zSet the total number of CPUs and GPUs used by training. If CPU or GPU resource limits are not set, they will be set to the total cluster resources minus the resources used by training. N)r r!)r"r&r'r#r#r$set_train_total_resources7s z$DataConfig.set_train_total_resourcesdatasets world_sizeworker_handlesworker_node_idsreturncKsddt|D}|D]\}}|jdur||q |jdkr(t|} nt|j} |jr2|nd} |D]M\} } t |j } | rS| j t|j|jd| _ | | } | | j_| | vrvt| j|d| dD] \}}|||| <qjq8t|D] }| ||| <qzq8|S)aConfigure how Train datasets should be assigned to workers. Args: datasets: The datasets dict passed to Train by the user. world_size: The number of Train workers in total. worker_handles: The actor handles of the Train workers. worker_node_ids: The node ids of the Train workers. kwargs: Forwards compatibility placeholder. Returns: A list of dataset splits for each worker. The size of the list must be equal to `world_size`. Each element of the list contains the assigned `DataIterator` instances by name for the worker. cSsg|]}iqSr#r#).0_r#r#r$ Xsz(DataConfig.configure..Nr)cpugpuT)equallocality_hints)rangeitemsnameset_namersetkeysrcopydeepcopyris_resource_limits_defaultexclude_resourcesaddr r r!contextr enumeratestreaming_splititerator)r"r)r*r+r,kwargsoutput dataset_namedatasetrr4r7dsrisplitr#r#r$ configureAs@        zDataConfig.configurecCs2tjj}t|jj|jj|jj|jj |jj dS)zThe default Ray Data options used for data ingest. By default, configurations are carried over from what is already set in DataContext. )locality_with_outputresource_limitsr>preserve_orderverbose_progress) raydata DataContext get_currentr rrLrMr>rNrO)ctxr#r#r$rs z!DataConfig.default_ingest_options)rNT)r __module__ __qualname____doc__rrrstrrr boolr%floatr(r rr intrr rrK staticmethodrr#r#r#r$r s6 $     @r)r;typingrrrrrrP ray.actorrray.datarr r r 9ray.data._internal.execution.interfaces.execution_optionsr ray.util.annotationsr rrr#r#r#r$s