o Me߱@sdZddlmZddlmZddlmZddlmZgdZ Gd d d e Z Gd d d e Z Gd dde Z Gddde ZGddde ZGddde ZdS)zBThis is definition of dataset class, which is high performance IO.) data_feed_pb2) text_format)core) deprecated)DatasetFactoryInMemoryDataset QueueDatasetc@s"eZdZdZddZdddZdS) ra^ DatasetFactory is a factory which create dataset by its name, you can create "QueueDataset" or "InMemoryDataset", or "FileInstantDataset", the default is "QueueDataset". Example: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") cCsdS) Init. Nselfr r DD:\Projects\ConvertPro\env\Lib\site-packages\paddle/fluid/dataset.py__init__&szDatasetFactory.__init__r cCs&z t|}|WStd|)a Create "QueueDataset" or "InMemoryDataset", or "FileInstantDataset", the default is "QueueDataset". Args: datafeed_class(str): datafeed class name, QueueDataset or InMemoryDataset. Default is QueueDataset. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() z datafeed class %s does not exist)globals ValueError)rZdatafeed_classdatasetr r rcreate_dataset*s zDatasetFactory.create_datasetN)r )__name__ __module__ __qualname____doc__rrr r r rrs rc@seZdZdZddZddZddZdd Zd,d d Zd dZ ddZ ddZ ddZ ddZ ddZddZddZddZdd Zd!d"Zd#d$Zd%d&Zd'd(Zd)d*Zd+S)- DatasetBasez Base dataset class. cCs:t|_d|j_td|_d|_g|_d|_ d|_ dS)r catZMultiSlotDatasetrFN) rZ DataFeedDesc proto_desc pipe_commandrDatasetr thread_numfilelist use_ps_gpupsgpur r r rrEs   zDatasetBase.__init__cC ||j_dS)a Set pipe command of current dataset A pipe command is a UNIX pipeline command that can be used only Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() dataset.set_pipe_command("python my_script.py") Args: pipe_command(str): pipe command N)rr)rrr r rset_pipe_commandQs zDatasetBase.set_pipe_commandcCr")aJ Set so parser name of current dataset Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() dataset.set_so_parser_name("./abc.so") Args: pipe_command(str): pipe command N)rso_parser_name)rr$r r rset_so_parser_namec zDatasetBase.set_so_parser_namecCr")aa Set rank_offset for merge_pv. It set the message of Pv. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() dataset.set_rank_offset("rank_offset") Args: rank_offset(str): rank_offset's name N)r rank_offset)rr'r r rset_rank_offsettr&zDatasetBase.set_rank_offsetTcCs|r |j||||_dS)a set fea eval mode for slots shuffle to debug the importance level of slots(features), fea_eval need to be set True for slots shuffle. Args: record_candidate_size(int): size of instances candidate to shuffle one slot fea_eval(bool): whether enable fea eval mode to enable slots shuffle. default is True. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_fea_eval(1000000, True) N)r set_fea_evalfea_eval)rZrecord_candidate_sizer*r r rr)s zDatasetBase.set_fea_evalcCs"|jrt|}|j|dSdSa Slots Shuffle Slots Shuffle is a shuffle method in slots level, which is usually used in sparse feature with large scale of instances. To compare the metric, i.e. auc while doing slots shuffle on one or several slots with baseline to evaluate the importance level of slots(features). Args: slots(list[string]): the set of slots(string) to do slots shuffle. Examples: import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_merge_by_lineid() #suppose there is a slot 0 dataset.slots_shuffle(['0']) N)r*setr slots_shufflerslotsZ slots_setr r rr-szDatasetBase.slots_shufflecCr")aG Set batch size. Will be effective during training Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() dataset.set_batch_size(128) Args: batch_size(int): batch size N)r batch_size)rr0r r rset_batch_sizer&zDatasetBase.set_batch_sizecCr")aW Set pv batch size. It will be effective during enable_pv_merge Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() dataset.set_pv_batch(128) Args: pv_batch_size(int): pv batch size N)r pv_batch_size)rr2r r rset_pv_batch_size zDatasetBase.set_pv_batch_sizecC|j|||_dS)a: Set thread num, it is the num of readers. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() dataset.set_thread(12) Args: thread_num(int): thread num N)rset_thread_numrrrr r r set_thread  zDatasetBase.set_threadcCr5)a@ Set file list in current worker. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() dataset.set_filelist(['a.txt', 'b.txt']) Args: filelist(list): file list N)r set_filelistr)rrr r rr:r9zDatasetBase.set_filelistcCs ||j_dSN)r input_type)rr<r r rset_input_types zDatasetBase.set_input_typecCs|jj}|D]D}|j}d|_|j|_|jdkr#d|_|j |j|j t j j jkr/d|_q|j t j j jkr;d|_q|j t j j jkrGd|_qtddS)a? Set Variables which you will use. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() dataset.set_use_var([data, label]) Args: var_list(list): variable list TrfloatZuint64Zuint32zQCurrently, fluid.dataset only supports dtype=float32, dtype=int32 and dtype=int64N)rZmulti_slot_descr/addZis_usednameZ lod_levelZis_denseshapeextendZdtyperZVarDescZVarTypeZFP32typeZINT64ZINT32r)rZvar_listZ multi_slotvarZslot_varr r r set_use_vars$  zDatasetBase.set_use_varcCs|j||dS)ae Set hdfs config: fs name ad ugi Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() dataset.set_hdfs_config("my_fs_name", "my_fs_ugi") Args: fs_name(str): fs name fs_ugi(str): fs ugi N)rset_hdfs_config)rZfs_nameZfs_ugir r rrFszDatasetBase.set_hdfs_configcC|j|dS)aa Set customized download cmd: download_cmd Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() dataset.set_download_cmd("./read_from_afs") Args: download_cmd(str): customized download command N)rset_download_cmd)rZ download_cmdr r rrH*szDatasetBase.set_download_cmdcCsH|jt|jkrt|j|_|j|j|j||jdS)h Set data_feed_desc before load or shuffle, user no need to call this function. N)rlenrrr6set_data_feed_descdesccreate_readersr r r r_prepare_to_run:s  zDatasetBase._prepare_to_runcCs,d|_ts d|_dS|jr||_dSdS)zQ set use_ps_gpu flag Args: use_ps_gpu: bool TFN)r r_is_compiled_with_heterpsr!)rr!r r r_set_use_ps_gpuEs   zDatasetBase._set_use_ps_gpucCs|jdSr;)rZdestroy_readersr r r r_finish_to_runSszDatasetBase._finish_to_runcCs t|jS)a7 Returns a protobuf message for this DataFeedDesc Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset() print(dataset.desc()) Returns: A string message )rZMessageToStringrr r r rrLVr4zDatasetBase.desccCdSr;r r7r r r_dynamic_adjust_before_trainfz(DatasetBase._dynamic_adjust_before_traincCrRr;r r r r r_dynamic_adjust_after_trainirTz'DatasetBase._dynamic_adjust_after_trainN)T)rrrrrr#r%r(r)r-r1r3r8r:r=rErFrHrNrPrQrLrSrUr r r rrBs,  !  rcs2eZdZdZedddfddZeddddd Zedd dd d Zedd dddZedddddZ edddddZ edddddZ edddddZ ddZ ddZedd dd!d"Zd#d$Zd%d&Zd'd(Zd)d*Zedd+dd`d-d.Zedd/ddad1d2Zedd3ddbd5d6Zedd7dd8d9Zedd:dd;d<Zd=d>Zedd?ddcdAdBZeddCddddEdFZeddGddHdIZeddJddKdLZeddMddedOdPZeddQddRdSZdTdUZ eddVddddWdXZ!eddYddddZd[Z"dcd\d]Z#d^d_Z$Z%S)fr z InMemoryDataset, it will load data into memory and shuffle data before training. This class should be created by DatasetFactory Example: dataset = paddle.fluid.DatasetFactory().create_dataset("InMemoryDataset") 2.0.0z"paddle.distributed.InMemoryDatasetZsinceZ update_tocs\tt|d|j_d|_d|_d|_d|_d|_ d|_ d|_ d|_ d|_ d|_d|_dS)r ZMultiSlotInMemoryDataFeedNFT)superr rrr@fleet_send_batch_sizeis_user_set_queue_num queue_num parse_ins_id parse_content parse_logkey merge_by_sidenable_pv_mergemerge_by_lineidfleet_send_sleep_seconds trainer_numr  __class__r rrws zInMemoryDataset.__init__z1paddle.distributed.InMemoryDataset._set_feed_typecCs(||j_|jjdkrtd|_dSdS)z$ Set data_feed_desc ZSlotRecordInMemoryDataFeedZSlotRecordDatasetN)rr@rrr)rZdata_feed_typer r r set_feed_types zInMemoryDataset.set_feed_typez2paddle.distributed.InMemoryDataset._prepare_to_runcCs|jdkrd|_|j|j|jdur|j|_|j|j|j|j|j|j|j |j |j |j |j |j|j||j|jdS)rIrrN)rrr6r\ set_queue_numset_parse_ins_idr]set_parse_contentr^set_parse_logkeyr_set_merge_by_sidr`set_enable_pv_mergerarKrLZcreate_channelrMr r r rrNs   zInMemoryDataset._prepare_to_runz?paddle.distributed.InMemoryDataset._dynamic_adjust_before_traincCs:|js|jr|j|dn|j|d|j|dSNTF)r[r rdynamic_adjust_channel_numdynamic_adjust_readers_numr7r r rrSs z,InMemoryDataset._dynamic_adjust_before_trainz>paddle.distributed.InMemoryDataset._dynamic_adjust_after_traincCs@|js|jr|j|jdn|j|jd|j|jdSrn)r[r rrorrpr r r rrUs z+InMemoryDataset._dynamic_adjust_after_trainz1paddle.distributed.InMemoryDataset._set_queue_numcCsd|_||_dS)au Set Dataset output queue num, training threads get data from queues Args: queue_num(int): dataset output queue num Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_queue_num(12) TN)r[r\)rr\r r rrhs zInMemoryDataset.set_queue_numz4paddle.distributed.InMemoryDataset._set_parse_ins_idcC ||_dS)a[ Set id Dataset need to parse insid Args: parse_ins_id(bool): if parse ins_id or not Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_parse_ins_id(True) N)r])rr]r r rri z InMemoryDataset.set_parse_ins_idz5paddle.distributed.InMemoryDataset._set_parse_contentcCrq)a` Set if Dataset need to parse content Args: parse_content(bool): if parse content or not Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_parse_content(True) N)r^)rr^r r rrjrrz!InMemoryDataset.set_parse_contentcCrq)a] Set if Dataset need to parse logkey Args: parse_content(bool): if parse logkey or not Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_parse_logkey(True) N)r_)rr_r r rrk z InMemoryDataset.set_parse_logkeycCrq)a8 Set trainer num Args: trainer_num(int): trainer num Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset._set_trainer_num(1) N)rd)rrdr r r_set_trainer_num rsz InMemoryDataset._set_trainer_numz4paddle.distributed.InMemoryDataset._set_merge_by_sidcCrq)au Set if Dataset need to merge sid. If not, one ins means one Pv. Args: merge_by_sid(bool): if merge sid or not Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_merge_by_sid(True) N)r`)rr`r r rrlrrz InMemoryDataset.set_merge_by_sidcCrq)ab Set if Dataset need to merge pv. Args: enable_pv_merge(bool): if enable_pv_merge or not Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_enable_pv_merge(True) N)ra)rrar r rrm2rsz#InMemoryDataset.set_enable_pv_mergecC|jdS)a Merge pv instance and convey it from input_channel to input_pv_channel. It will be effective when enable_pv_merge_ is True. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.load_into_memory() dataset.preprocess_instance() N)rpreprocess_instancer r r rrvCsz#InMemoryDataset.preprocess_instancecCrG)a Set current phase in train. It is useful for untest. current_phase : 1 for join, 0 for update. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.load_into_memory() dataset.set_current_phase(1) N)rset_current_phase)rZ current_phaser r rrwUsz!InMemoryDataset.set_current_phasecCru)a Divide pv instance and convey it to input_channel. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.load_into_memory() dataset.preprocess_instance() exe.train_from_dataset(dataset) dataset.postprocess_instance() N)rpostprocess_instancer r r rrxgz$InMemoryDataset.postprocess_instancez=paddle.distributed.InMemoryDataset._set_fleet_send_batch_sizecCrq)ar Set fleet send batch size, default is 1024 Args: fleet_send_batch_size(int): fleet send batch size Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_fleet_send_batch_size(800) N)rZ)rrZr r rset_fleet_send_batch_sizez z)InMemoryDataset.set_fleet_send_batch_sizez@paddle.distributed.InMemoryDataset._set_fleet_send_sleep_secondsrcCrq)as Set fleet send sleep time, default is 0 Args: fleet_send_sleep_seconds(int): fleet send sleep time Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_fleet_send_sleep_seconds(2) N)rc)rrcr r rset_fleet_send_sleep_secondsr|z,InMemoryDataset.set_fleet_send_sleep_secondsz7paddle.distributed.InMemoryDataset._set_merge_by_lineidrcCs|j|d|_d|_dS)a Set merge by line id, instances of same line id will be merged after shuffle, you should parse line id in data generator. Args: merge_size(int): ins size to merge. default is 2. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_merge_by_lineid() TN)rset_merge_by_lineidrbr])rZ merge_sizer r rr~s  z#InMemoryDataset.set_merge_by_lineidz@paddle.distributed.InMemoryDataset._set_generate_unique_feasignscCs|j|||_||_dSr;)rset_generate_unique_feasignsZgen_uni_feasignsZlocal_shard_num)rZgenerate_uni_feasigns shard_numr r rrs  z,InMemoryDataset.set_generate_unique_feasignsz@paddle.distributed.InMemoryDataset._generate_local_tables_unlockcCs|j|||||dSr;)rgenerate_local_tables_unlock)rZtable_idZfea_dimZread_thread_numZconsume_thread_numrr r rrs z,InMemoryDataset.generate_local_tables_unlockcCsZt|dd}t|dd}t|dd}|jr)tr+|j|||dSdSdS)a :api_attr: Static Graph Set training date for pull sparse parameters, saving and loading model. Only used in psgpu Args: date(str): training date(format : YYMMDD). eg.20211111 Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") dataset.set_date("20211111") N)intr rrOr!set_daterdateyearmonthdayr r rrs zInMemoryDataset.set_datez3paddle.distributed.InMemoryDataset.load_into_memoryFcCsF||js|jdStr!|j|j|j|dSdS)a Load data into memory Args: is_shuffle(bool): whether to use local shuffle, default is False Examples: .. code-block:: python # required: skiptest import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.load_into_memory() N)rNr rload_into_memoryrrOr!Z set_dataset)rZ is_shuffler r rrsz InMemoryDataset.load_into_memoryz6paddle.distributed.InMemoryDataset.preload_into_memoryNcCs:||dur |j}|j||j|jdS)a Load data into memory in async mode Args: thread_num(int): preload thread num Examples: .. code-block:: python # required: skiptest import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.preload_into_memory() dataset.wait_preload_done() N)rNrrZset_preload_thread_numZcreate_preload_readerspreload_into_memoryr7r r rrs   z#InMemoryDataset.preload_into_memoryz4paddle.distributed.InMemoryDataset.wait_preload_donecCs|j|jdS)a Wait preload_into_memory done Examples: .. code-block:: python # required: skiptest import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.preload_into_memory() dataset.wait_preload_done() N)rwait_preload_doneZdestroy_preload_readersr r r rrs z!InMemoryDataset.wait_preload_donez0paddle.distributed.InMemoryDataset.local_shufflecCru)a Local shuffle Examples: .. code-block:: python # required: skiptest import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.load_into_memory() dataset.local_shuffle() N)r local_shuffler r r rr0ryzInMemoryDataset.local_shufflez1paddle.distributed.InMemoryDataset.global_shuffle cCs.|dur!t|drtd|n|j|jdkr!||_|jdur)d|_|jdur1d|_|j |j |j|j |j|j |j|dur^t|drY|n|j|j ||durwt|drr|n|j|jr|j|durt|dr|dS|jdSdS)ag Global shuffle. Global shuffle can be used only in distributed mode. i.e. multiple processes on single machine or multiple machines training together. If you run in distributed mode, you should pass fleet instead of None. Examples: .. code-block:: python # required: skiptest import paddle.fluid as fluid from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.load_into_memory() dataset.global_shuffle(fleet) Args: fleet(Fleet): fleet singleton. Default None. thread_num(int): shuffle thread num. Default is 12. Nbarrier_workerz pscore fleetrXrzr)hasattrprintr _role_makerrdZ worker_numrZrcrZ"register_client2client_msg_handlerZset_trainer_numr{r}global_shufflerb)rfleetrr r rrCs>                  zInMemoryDataset.global_shufflez1paddle.distributed.InMemoryDataset.release_memorycCru)a8 :api_attr: Static Graph Release InMemoryDataset memory data, when data will not be used again. Examples: .. code-block:: python # required: skiptest import paddle.fluid as fluid from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.load_into_memory() dataset.global_shuffle(fleet) exe = fluid.Executor(fluid.CPUPlace()) exe.run(fluid.default_startup_program()) exe.train_from_dataset(fluid.default_main_program(), dataset) dataset.release_memory() N)rrelease_memoryr r r rrszInMemoryDataset.release_memorycCs |jS)a Get memory data size of Pv, user can call this function to know the pv num of ins in all workers after load into memory. Note: This function may cause bad performance, because it has barrier Returns: The size of memory pv data. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.load_into_memory() print dataset.get_pv_data_size() )rget_pv_data_sizer r r rrs z InMemoryDataset.get_pv_data_sizez7paddle.distributed.InMemoryDataset.get_memory_data_sizecCsLddl}|j}||g}|dur"|d}|j|||dS|dS)a6 Get memory data size, user can call this function to know the num of ins in all workers after load into memory. Note: This function may cause bad performance, because it has barrier Args: fleet(Fleet): Fleet Object. Returns: The size of memory data. Examples: .. code-block:: python # required: skiptest import paddle.fluid as fluid from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.load_into_memory() print dataset.get_memory_data_size(fleet) rN)numpyrget_memory_data_sizearrayrall_reduce_workerrrnpZlocal_data_sizeZglobal_data_sizer r rrs  z$InMemoryDataset.get_memory_data_sizez8paddle.distributed.InMemoryDataset.get_shuffle_data_sizecCstddl}|j}||g}td||dur6|d}t|dr+|j|}|dS|j |||dS|dS)a Get shuffle data size, user can call this function to know the num of ins in all workers after local/global shuffle. Note: This function may cause bad performance to local shuffle, because it has barrier. It does not affect global shuffle. Args: fleet(Fleet): Fleet Object. Returns: The size of shuffle data. Examples: .. code-block:: python # required: skiptest import paddle.fluid as fluid from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.load_into_memory() dataset.global_shuffle(fleet) print dataset.get_shuffle_data_size(fleet) rNz global shuffle local_data_size: util) rrget_shuffle_data_sizerrrrZ all_reducerrrr r rrs     z%InMemoryDataset.get_shuffle_data_sizecCrG)zO Set heter ps mode user no need to call this function. N)rZ set_heter_ps)rZenable_heter_psr r r _set_heter_ps szInMemoryDataset._set_heter_pscCs|dd|jj_|dd|jj_|dd|jj_|dd|jj_|d d |jj_|d d|jj_|d d |jj_ |dd|jj_ |dd|jj_ |dd|jj_ |j ddS)a Set graph config, user can set graph config in gpu graph mode. Args: config(dict): config dict. Returns: The size of shuffle data. Examples: .. code-block:: python # required: skiptest import paddle.fluid as fluid from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset") graph_config = {"walk_len": 24, "walk_degree": 10, "once_sample_startid_len": 80000, "sample_times_one_chunk": 5, "window": 3, "debug_mode": 0, "batch_size": 800, "meta_path": "cuid2clk-clk2cuid;cuid2conv-conv2cuid;clk2cuid-cuid2clk;clk2cuid-cuid2conv", "gpu_graph_training": 1} dataset.set_graph_config(graph_config) walk_degreerwalk_lenwindowonce_sample_startid_leni@sample_times_one_chunk r0 debug_moderfirst_node_type meta_pathgpu_graph_trainingTN)getrZ graph_configrrrrrr0rrrrrZset_gpu_graph_mode)rconfigr r rset_graph_configs&    z InMemoryDataset.set_graph_config)rz)r)r)Fr;)Nr)&rrrrrrrgrNrSrUrhrirjrkrtrlrmrvrwrxr{r}r~rrrrrrrrrrrrrr __classcell__r r rerr ms                   ;  %  +r csFeZdZdZfddZedddddZd d Zdd d ZZ S)r z QueueDataset, it will process data streamly. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("QueueDataset") ctt|d|j_dS)z` Initialize QueueDataset This class should be created by DatasetFactory ZMultiSlotDataFeedN)rYr rrr@r rer rrJ zQueueDataset.__init__rVz/paddle.distributed.QueueDataset._prepare_to_runrWcCsf|jt|jkrt|j|_|jdkrd|_|j|j|j|j|j||jdS)zp Set data_feed_desc/thread num/filelist before run, user no need to call this function. rrN) rrJrrr6r:rKrLrMr r r rrNRs  zQueueDataset._prepare_to_runcCtd)a Local shuffle data. Local shuffle is not supported in QueueDataset NotImplementedError will be raised Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("QueueDataset") dataset.local_shuffle() Raises: NotImplementedError: QueueDataset does not support local shuffle zYQueueDataset does not support local shuffle, please use InMemoryDataset for local_shuffleNotImplementedErrorr r r rrbszQueueDataset.local_shuffleNcCr)aa Global shuffle data. Global shuffle is not supported in QueueDataset NotImplementedError will be raised Args: fleet(Fleet): fleet singleton. Default None. Examples: .. code-block:: python import paddle.fluid as fluid from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet dataset = fluid.DatasetFactory().create_dataset("QueueDataset") dataset.global_shuffle(fleet) Raises: NotImplementedError: QueueDataset does not support global shuffle z[QueueDataset does not support global shuffle, please use InMemoryDataset for global_shufflerrrr r rrxszQueueDataset.global_shuffler;) rrrrrrrNrrrr r rerr >s  r cs2eZdZdZfddZddZd ddZZS) FileInstantDatasetz FileInstantDataset, it will process data streamly. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory.create_dataset("FileInstantDataset") cr)zf Initialize FileInstantDataset This class should be created by DatasetFactory ZMultiSlotFileInstantDataFeedN)rYrrrr@r rer rrrzFileInstantDataset.__init__cCr)zY Local shuffle FileInstantDataset does not support local shuffle z_FileInstantDataset does not support local shuffle, please use InMemoryDataset for local_shufflerr r r rrz FileInstantDataset.local_shuffleNcCr)z[ Global shuffle FileInstantDataset does not support global shuffle zaFileInstantDataset does not support global shuffle, please use InMemoryDataset for global_shufflerrr r rrrz!FileInstantDataset.global_shuffler;)rrrrrrrrr r rerrs   rcsheZdZdZfddZddZddZdd Zd d Zd d Z ddZ ddZ ddZ ddZ ZS) BoxPSDatasetz BoxPSDataset: derived from InMemoryDataset. Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset") cs(tt|t|j|_d|j_dS)z` Initialize BoxPSDataset This class should be created by DatasetFactory ZPaddleBoxDataFeedN) rYrrrZBoxPSrboxpsrr@r rer rrs zBoxPSDataset.__init__cCsDt|dd}t|dd}t|dd}|j|||dS)z% Workaround for date Nrr)rrrrr r rrszBoxPSDataset.set_datecCru)a: Begin Pass Notify BoxPS to load sparse parameters of next pass to GPU Memory Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset") dataset.begin_pass() N)r begin_passr r r rrs zBoxPSDataset.begin_passcCrG)a End Pass Notify BoxPS that current pass ended Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset") dataset.end_pass(True) N)rend_pass)rZneed_save_deltar r rrs zBoxPSDataset.end_passcCru)a Wait async preload done Wait Until Feed Pass Done Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.preload_into_memory() dataset.wait_preload_done() N)rZwait_feed_pass_doner r r rrszBoxPSDataset.wait_preload_donecC||jdS)a Load next pass into memory and notify boxps to fetch its emb from SSD Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.load_into_memory() N)rNrrr r r rr zBoxPSDataset.load_into_memorycCr)a Begin async preload next pass while current pass may be training Examples: .. code-block:: python import paddle.fluid as fluid dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset") filelist = ["a.txt", "b.txt"] dataset.set_filelist(filelist) dataset.preload_into_memory() N)rNrrr r r rrrz BoxPSDataset.preload_into_memorycCs$|js |j|d|j|dS)NT)r[rrorpr7r r rrSsz)BoxPSDataset._dynamic_adjust_before_traincCrRr;r r r r rrU$rTz(BoxPSDataset._dynamic_adjust_after_traincCst|}|j|dSr+)r,rr-r.r r rr-'szBoxPSDataset.slots_shuffle)rrrrrrrrrrrrSrUr-rr r rerrs    rN)rZpaddle.fluid.protorZgoogle.protobufrrrutilsr__all__objectrrr r rrr r r rs&    *-VU&