o "j^@sddlZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z gZ Gddde ZGddde ZGd d d e ZGd d d e ZGd ddeZGdddZGdddeZdddZGdddeZGdddeZdS)N)core)loggerc@ eZdZdS) ExecuteErrorN__name__ __module__ __qualname__r r b/var/www/html/Deteccion_Ine/venv/lib/python3.10/site-packages/paddle/distributed/fleet/utils/fs.pyr rc@r)FSFileExistsErrorNrr r r r r$r rc@r)FSFileNotExistsErrorNrr r r r r(r rc@r) FSTimeOutNrr r r r r,r rc@r)FSShellCmdAbortedNrr r r r r0r rc@seZdZejddZejddZejddZejddZejd d Z ejd d Z ejd dZ ejddZ ejddZ ejddZejd"ddZejddZejddZejd#ddZejd$d d!ZdS)%FScCtNNotImplementedErrorselffs_pathr r r ls_dir5z FS.ls_dircCrrrrr r r is_file9rz FS.is_filecCrrrrr r r is_dir=rz FS.is_dircCrrrrr r r is_existArz FS.is_existcCrrr)r local_pathrr r r uploadErz FS.uploadcCrrr)rrrr r r downloadIrz FS.downloadcCrrrrr r r mkdirsMrz FS.mkdirscCrrrrr r r deleteQrz FS.deletecCrrrrr r r need_upload_downloadUrzFS.need_upload_downloadcCrrrr fs_src_path fs_dst_pathr r r renameYrz FS.renameFcCrrrrr'r( overwrite test_existsr r r mv]rzFS.mvcCrrr)r local_dirdest_dirr r r upload_dirarz FS.upload_dircCrrrrr r r list_dirserz FS.list_dirsTcCrrrrrexist_okr r r touchirzFS.touchNcCrrrrr r r catmrzFS.catFFTr)rr r abcabstractmethodrrrrr r!r"r#r%r)r-r0r1r4r5r r r r r4s>              rc@s|eZdZdZddZddZddZdd Zd d Zd d Z ddZ ddZ ddZ ddZ dddZd ddZddZdS)!LocalFSa( A tool of local file system. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> client = LocalFS() >>> subdirs, files = client.ls_dir("./") cCs\||s ggfSg}g}t|D]}tj|d|r$||q||q||fS)az List directorys and files under `fs_path` . Args: fs_path(str): The local file path. Returns: Tuple: Return a 2-tuple, the first is a list of all its subdirectories, and the second is a list of all its subfiles, e.g. ([subdirname1, subdirname1, ...], [filename1, filename2, ...]). Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> client = LocalFS() >>> subdirs, files = client.ls_dir("./") /)roslistdirpathisdirappend)rrdirsfilesfr r r rs   zLocalFS.ls_dircCs,tj|r J|dtj|dddS)a Create a local directory. Args: fs_path(str): The local directory path. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> client = LocalFS() >>> client.mkdirs("test_mkdirs") >>> client.delete("test_mkdirs") z is already a fileT)r3N)r<r>isfilemakedirsrr r r r"szLocalFS.mkdirscCst||dS)aI Rename the file. Args: fs_src_path(str): The actual name of the file or directory fs_dst_path(str): The new name of the file or directory. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> client = LocalFS() >>> client.touch("test_rename_src") >>> print(client.is_exist("test_rename_src")) True >>> client.rename("test_rename_src", "test_rename_dst") >>> print(client.is_exist("test_rename_src")) False >>> print(client.is_exist("test_rename_dst")) True >>> client.delete("test_rename_dst") N)r<r)r&r r r r)szLocalFS.renamecCt|dSr)shutilrmtreerr r r _rmrz LocalFS._rmrcCrFr)r<removerr r r _rmrJz LocalFS._rmcCs.||sdStj|r||S||S)a Delete the local file path, whether it's a file or directory. Args: fs_path(str): The local file path. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> client = LocalFS() >>> client.mkdirs("test_localFS_mkdirs") >>> client.delete("test_localFS_mkdirs") N)rr<r>rDrLrIrr r r r#s    zLocalFS.deletecCdS)NFr r$r r r r%zLocalFS.need_upload_downloadcC tj|S)au Whether the local file path is a file. Args: fs_path(str): The local file path. Returns: Bool: Return true if the path exists and it's a file, otherwise return false. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> client = LocalFS() >>> client.touch("test_is_file") >>> print(client.is_file("test_is_file")) True >>> client.delete("test_is_file") )r<r>rDrr r r r zLocalFS.is_filecCrO)a| Whether the local file path is a directory. Args: fs_path(str): The local file path. Returns: Bool: Return true if the path exists and it's a directory, otherwise return false. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> client = LocalFS() >>> client.mkdirs("test_is_dir") >>> print(client.is_dir("test_is_dir")) True >>> client.delete("test_is_dir") r<r>r?rr r r rrPzLocalFS.is_dircCrO)a Whether the local file path exists. Args: fs_path(str): The local file path. Returns: Bool: Wheter it's a file or directory, return true if the path exists, otherwise return false. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> local_fs = LocalFS() >>> ret = local_fs.is_exist("test_is_exist") )r<r>existsrr r r r)s zLocalFS.is_existTcCsF||r |r dStt|d WddS1swYdS)a1 Create a local file. Args: fs_path(str): The local file path. exist_ok(bool): When `fs_path` exists, if `exist_ok` is set false, program will throw an Exception. Default is true. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> client = LocalFS() >>> client.touch("test_touch") >>> client.delete("test_touch") Na)rropenr2r r r r4@s  "z LocalFS.touchFcCs@||st|r||r||||rt|||S)a Move a local file or directory from `src_path` to `dst_path` . Args: src_path(str): Name of the file or directory, that's needed to be moved. dst_path(str): Name of the file or directory to which to move to. overwrite(bool): Whether to re-write `dst_path` if that exists. Default is False. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> client = LocalFS() >>> client.touch("test_mv_src") >>> client.mv("test_mv_src", "test_mv_dst") >>> client.delete("test_mv_dst") )rrr#rr))rZsrc_pathZdst_pathr+r,r r r r-[s    z LocalFS.mvcs*|sgSfddtD}|S)a Only list directorys under `fs_path` . Args: fs_path(str): The local file path. Returns: List: A list of all its subdirectories, e.g. [subdirname1, subdirname1, ...]. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> client = LocalFS() >>> subdirs = client.list_dirs("./") cs$g|]}tjd|r|qSr;rQ).0rCrr r s z%LocalFS.list_dirs..)rr<r=)rrrAr rWr r1{s  zLocalFS.list_dirsNr7r6)rr r __doc__rr"r)rIrLr#r%rrrr4r-r1r r r r r:rs"   r:csfdd}|S)Ncstfdd}|S)Nc s|d}}|durt|jd}n|d}t|jd}t}|} z|i|WStyY}z!t||krJtd|dt|t|WYd}~nd}~wwt|dkrrtd|t|t}q$)Nrg@@Tzargs:z timeout:z*hadoop operator timeout:args:{} timeout:{}) float _time_out _sleep_intertimerrsleepprintformat)argskwargsotime_outinterstartZlast_print_timee)rC max_time_outr r handlers6 z2_handle_errors..decorator..handler) functoolswraps)rCrjri)rCr decoratorsz!_handle_errors..decoratorr )rirnr rmr _handle_errorss #roc@sPeZdZdZ  dBddZdCdd ZdCd d Zed d ZeddZ ddZ ddZ eddZ ddZ ddZeddZdDddZdEddZed d!ZdEd"d#Zed$d%Zed&d'ZdFd)d*Zed+d,Zd-d.Zd/d0Zed1d2ZdGd3d4Zed5d6Zd7d8ZdHd:d;Zedd?Z d@dAZ!d9S)I HDFSClienta A tool of HDFS. Args: hadoop_home(str): Hadoop home. configs(dict): Hadoop config. It is a dictionary and needs to contain the keys: "fs.default.name" and "hadoop.job.ugi". Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> client.ls_dir("hdfs:/test_hdfs_client") ([], []) c Csg|_d|}|j|d}|j||r.|D]\}}d|d|} |j| q||_||_d|j|_td|_ dS)Nz %s/bin/hadoopfsz-D= z8\s?responseErrorMsg\s?\:.*, errorCode\:\s?[0-9]+, path\:) Z pre_commandsr@itemsr\r]join _base_cmdrecompile _bd_err_re) rZ hadoop_homeZconfigsre sleep_interZ hadoop_bindfskvZconfig_commandr r r __init__s   zHDFSClient.__init__Fc Cs||jd|}d}d}d}t|dD]}t|dd|\}}t|}|dkr*nt|q|dkr8t|||fS)Nz -rr) rxrangerZshell_execute_cmdintr^r_r splitlines) rcmdredirect_stderr retry_timesexe_cmdretoutputretry_sleep_secondxr r r _run_cmds  zHDFSClient._run_cmdc Cs|jg|}d}d}d}t|dD]I}ztj|dtj|r"tjntjdd} | j}Wn0tjyK} z| j }| j }t |WYd} ~ qd} ~ wt y]} zWYd} ~ nd} ~ ww|dkrft|dS)NrrrT)checkstdoutstderrtextr)rxsplitr subprocessrunPIPESTDOUTrCalledProcessError returncoderr^r_ Exceptionr) rrrrrrrrrprocessrhr r r _run_safe_cmds: zHDFSClient._run_safe_cmdcC ||sgS||\}}|S)a) Only list directorys under `fs_path` . Args: fs_path(str): The HDFS file path. Returns: List: A list of all its subdirectories, e.g. [subdirname1, subdirname1, ...]. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> subdirs = client.list_dirs("hdfs:/test_hdfs_client") r_ls_dirrrrArBr r r r1!s zHDFSClient.list_dirscC||s ggfS||S)a List directorys and files under `fs_path` . Args: fs_path(str): The HDFS file path. Returns: Tuple: Return a 2-tuple, the first element is the list of all its subdirectories, and the second one is the list of all its subfiles, e.g. ([subdirname1, subdirname1, ...], [filename1, filename2, ...]). Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> subdirs, files = client.ls_dir("hdfs:/test_hdfs_client") rrr r r rCs  zHDFSClient.ls_dirc Csd|g}||\}}|dkrt|g}g}|D](}|}t|dkr&qtj|d} |dddkr<|| q|| q||fS)Nz-lsrd)rrrlenr<r>basenamer@) rrrrlinesrArBlinearrpr r r res   zHDFSClient._ls_dircCs*|D]}|j|}|dur|SqdSr)r{match)rrlmr r r _test_match{s  zHDFSClient._test_matchcC||sdS||S)a. Whether the remote HDFS path is a directory. Args: fs_path(str): The HDFS file path. Returns: Bool: Return true if the path exists and it's a directory, otherwise return false. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> ret = client.is_file("hdfs:/test_hdfs_client") Fr_is_dirrr r r rs  zHDFSClient.is_dircCsRd|}|j|ddd\}}|r'||r%tdtd|t|dSdS)Nztest -d Trrrzraise exception:  F)rrr`rwr)rrrrrr r r rs  zHDFSClient._is_dircC||sdS|| S)a$ Whether the remote HDFS path is a file. Args: fs_path(str): The HDFS file path. Returns: Bool: Return true if the path exists and it's a file, otherwise return false. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> ret = client.is_file("hdfs:/test_hdfs_client") Frrr r r rs  zHDFSClient.is_filecCs0d|d}|j|ddd\}}|dkrdSdS)aB Whether the remote HDFS path exists. Args: fs_path(str): The hdfs file path. Returns: Bool: Whether it's is file or directory, return true if the path exists, otherwise return false. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DITSTRIBUTED) >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> ret = client.is_exist("hdfs:/test_hdfs_client") ztest -e ruTrrrF)r)rrrroutr r r rs zHDFSClient.is_existcCsl|d}|d}tj|}||d|r$|r$||d|||s.|||||dSz upload dir to hdfs Args: local_dir(str): local dir dest_dir(str): hdfs dest dir overwrite(bool): is overwrite Returns: return code r;N)rstripr<r>rrr#r" _try_uploadrr.r/r+Zlocal_basenamer r r r0s    zHDFSClient.upload_dircsfdd}dd}t}||st|d||}|s%tddS|r6|r6||g} t|D]} || |} tj ||| fd} | | | q<| D]} | qZdS) a Upload the local path to remote HDFS. Args: local_path(str): The local path. fs_path(str): The HDFS path. multi_processes(int|1): the upload data process at the same time, default=5 overwrite(bool|False): will overwrite file on HDFS or not Examples: .. code-block:: python >>> # doctest: +SKIP('depend on external file') >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> client.upload("test_hdfs_client", "hdfs:/test_hdfs_client") cs|D]}||qdSr)r)Zhdfs_path_singledatasdatar$r r Z__subprocess_upload&sz.HDFSClient.upload..__subprocess_uploadcSsZg}tj|s |Stj|r&t|D]}tj||}||q|S|||S)z get local files Args: path(str): local path Returns: list of local files )r<r>rRr?r=rwr@)r>Zrlistfiletr r r get_local_files*s    z*HDFSClient.upload..get_local_files not existsz/there are nothing need to upload, function exitNtargetrb) r:rrr`r#r"r _split_filesmultiprocessingProcessr@rgrw)rrrmulti_processesr+Z_HDFSClient__subprocess_uploadrlocal all_filesprocsi process_datasrprocr r$r r s.        zHDFSClient.uploadc Cs`d|d|}d}z||\}}|dkrt|WdSty/}z|||d}~ww)Nzput rur)rrrr#)rrrrr_rhr r r rYs zHDFSClient._try_uploadcsfdd}stdr|S\}}fdd|D}|fdd|Dg} t|D]} || |} tj ||| fd} | | | q?| D]} | q]dS) al Download remote HDFS path to the local. Args: fs_path(str): The HDFS path. local_path(str): The local path. multi_processes(int|1): the download data process at the same time, default=1 overwrite(bool): is overwrite Examples: .. code-block:: python >>> # doctest: +SKIP('depend on external file') >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> client.download("hdfs:/test_hdfs_client", "./") cs|D]}||qdSz download file from HDFS Args: local_path(str): the local file path datas(str): the hdfs file path list N) _try_downloadrrrr$r r __subprocess_downloadsz2HDFSClient.download..__subprocess_download not exitscg|]}d|qSrUr rVrrWr r rXz'HDFSClient.download..crrUr rrWr r rXrrN) rrrrrextendrrrrr@rgrw)rrrrr+Z _HDFSClient__subprocess_downloadrA all_filenamesrrrrrrr rrr r!fs&        zHDFSClient.downloadc Csfd|d|}d}z||\}}|dkrt|WdSty2}z t}|||d}~ww)Nzget rur)rrrr:r#)rrrrrrrhZlocal_fsr r r rs zHDFSClient._try_downloadcCs||rdSd}d|d}|j|dd\}}|dkr/|D] }d|vr(d}nq|s/t||rJ||sLd |}||\}}|dkrNt|dSdSdS) a Create a remote HDFS directory. Args: fs_path(str): The HDFS directory path. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on external file') >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> client.mkdirs("hdfs:/test_hdfs_client") NFzmkdir ruT)rrzNo such file or directoryz mkdir -p )rrr)rrZout_hdfsrrrrrr r r r"s(   zHDFSClient.mkdirsTcCsX|r ||r |||r&||st|d||r&t|d|||S)a Move a remote HDFS file or directory from `fs_src_path` to `fs_dst_path` . Args: fs_src_path(str): Name of the file or directory, that's needed to be moved. fs_dst_path(str): Name of the file or directory to which to move to. overwrite(bool): Whether to re-write `fs_dst_path` if that exists. Default is False. test_exists(bool): Check the existence of `fs_src_path` and `fs_dst_path` . When `test_exists` is set true, if `fs_src_path` doesn't exist or `fs_dst_path` exists, program will throw an Excetption. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on external file') >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> client.mv("hdfs:/test_hdfs_client", "hdfs:/test_hdfs_client2")  is not exists exists already)rr#rr_try_mvr*r r r r-s    z HDFSClient.mvc Cs|d|d|}d}z|j|dd\}}|dkrt|WdSty=}z||s7||r7WYd}~dS|d}~ww)Nzmv rurrr)rrrr)rr'r(rrrrhr r r rszHDFSClient._try_mvcC,d|}||\}}|dkrt|dS)Nzrmr rrrrrrrrr r r rI zHDFSClient._rmrcCr)Nzrm rrrr r r rLrzHDFSClient._rmcCs0||sdS||}|r||S||S)a Delete a remote HDFS path, whether it's a file or directory. Args: fs_path(str): The HDFS file path. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> client.delete("hdfs:/test_hdfs_client") N)rrrIrL)rrrr r r r#s    zHDFSClient.deletecCs ||r |r dSt||S)a6 Create a remote HDFS file. Args: fs_path(str): The HDFS file path. exist_ok(bool): When `fs_path` exists, if `exist_ok` is set false, program will throw an Exception. Default is true. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on external file') >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> client.touch("hdfs:/test_hdfs_client") N)rr_touchzr2r r r r4?s  zHDFSClient.touchcCr)Nztouchz rrrr r r r`s zHDFSClient._touchzcCrMNTr r$r r r r%grNzHDFSClient.need_upload_downloadNcCs"||r||}d|SdS)a Cat a remote HDFS file. Args: fs_path(str): The HDFS file path. Returns: file content Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import HDFSClient >>> hadoop_home = "/home/client/hadoop-client/hadoop/" >>> configs = { ... "fs.default.name": "hdfs://xxx.hadoop.com:54310", ... "hadoop.job.ugi": "hello,hello123" ... } >>> client = HDFSClient(hadoop_home, configs) >>> client.cat("hdfs:/test_hdfs_client") '' rr)r_try_catrw)rrrr r r r5js   zHDFSClient.catcCs0d|}|j|dd\}}|dkrt||S)Nzcat rrrr)rrrrrr r r rs zHDFSClient._try_catc Ct||}t||}|g|}t|D] }||d7<qgg|}d} t|D]}|| | ||||<| ||7} q+||Sz split file list Args: files(list): file list trainer_id(int): trainer mpi rank id trainers(int): all trainers num Returns: fileist(list): file list of current trainer rrrr rrBZ trainer_idZtrainers remainder blocksizeblocksrZ trainer_filesbeginr r r r     zHDFSClient._split_filesc Cst|dkrgSg}d}|D]}||d7}qd|d}||\}}t|dkr3td|gS|D] }|d} t| dkrCq5| d} t| d} || | d q5|S) z list_files return file path and size Args: path_list(list): file list Returns: fileist(list): file list with file path and size rrruzls z) | awk '{if ($8 != "") {print $5" "$8 }}'zlist_files empty, path[%s]r)r>size)rrrwarningrrr@) r path_list file_listZ str_concatr>rrrrr file_path file_sizer r r list_files_infos(      zHDFSClient.list_files_inforqrr)FrF)rFFTr7r)"rr r rYrrrror1rrrrrrrr0r rr!rr"r-rrIrLr#r4rr%r5rrrr r r r rpsX   ! !    # O  <   -'    !!  "  rpc@seZdZdZd.ddZddZdd Zd d Zd d ZddZ ddZ ddZ ddZ d/ddZ d0ddZd0ddZddZd1d!d"Zd#d$Zd2d%d&Zd'd(Zd3d*d+Zd,d-Zd)S)4 AFSClienta A tool of AFS. Use AfsWrapper. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> client.ls_dir("hdfs:/test_hdfs_client") rqrrcCst|_||_dSr)rZ AfsWrapper_fsr\)rrer|r r r rs  zAFSClient.__init__cCs|j||||dSr)rinit)rZfs_nameZfs_userZ fs_passwdZfs_confr r r rszAFSClient.initcCr)aw Only list directorys under `fs_path` . Args: fs_path(str): The HDFS file path. Returns: List: A list of all its subdirectories, e.g. [subdirname1, subdirname1, ...]. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> subdirs = client.list_dirs("hdfs:/test_hdfs_client") rrr r r r1s zAFSClient.list_dirscCr)a  List directorys and files under `fs_path` . Args: fs_path(str): The HDFS file path. Returns: Tuple: Return a 2-tuple, the first element is the list of all its subdirectories, and the second one is the list of all its subfiles, e.g. ([subdirname1, subdirname1, ...], [filename1, filename2, ...]). Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> subdirs, files = client.ls_dir("hdfs:/test_hdfs_client") rrr r r rs  zAFSClient.ls_dircCs|j|}|g}||fSr)rlist)rrrBrAr r r r s zAFSClient._ls_dircCr)a{ Whether the remote HDFS path is a directory. Args: fs_path(str): The HDFS file path. Returns: Bool: Return true if the path exists and it's a directory, otherwise return false. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> ret = client.is_dir("hdfs:/test_hdfs_client") Frrr r r r%s  zAFSClient.is_dircCs |j|}t|dkrdSdS)NrTF)rrr)rr list_pathr r r r@s  zAFSClient._is_dircCr)ar Whether the remote HDFS path is a file. Args: fs_path(str): The HDFS file path. Returns: Bool: Return true if the path exists and it's a file, otherwise return false. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> ret = client.is_file("hdfs:/test_hdfs_client") Frrr r r rGs  zAFSClient.is_filecCs |j|S)a Whether the remote HDFS path exists. Args: fs_path(str): The hdfs file path. Returns: Bool: Whether it's is file or directory, return true if the path exists, otherwise return false. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> ret = client.is_exist("hdfs:/test_hdfs_client") )rexistrr r r rbrPzAFSClient.is_existFcCsn|d}|d}tj|}||d|r$|r$||d|||s.|||j||dSr) rr<r>rrr#r"rr rr r r r0{s    zAFSClient.upload_dirrcCs0t}||st|d|j||dS)a Upload the local path to remote HDFS. Args: local_path(str): The local path. fs_path(str): The HDFS path. multi_processes(int|1): the upload data process at the same time, default=5 overwrite(bool|False): will overwrite file on HDFS or not Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> client.upload("test_hdfs_client", "hdfs:/test_hdfs_client") rN)r:rrrr )rrrrr+rr r r r s zAFSClient.uploadcsfdd}stdrj|S\}}fdd|D}g} t|D]} || |} tj ||| fd} | | | q4| D]} | qRdS)a Download remote HDFS path to the local. Args: fs_path(str): The HDFS path. local_path(str): The local path. multi_processes(int|1): the download data process at the same time, default=1 overwrite(bool): is overwrite Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> client.download("hdfs:/test_hdfs_client", "./") cs|D] }j||qdSr)rr!rr$r r rsz1AFSClient.download..__subprocess_downloadrcsg|]}|qSr r rrWr r rXsz&AFSClient.download..rN) rrrrr!rrrrrr@rgrw)rrrrr+Z_AFSClient__subprocess_downloadrrrrrrrrr rr r!s$       zAFSClient.downloadcCs||rdS|j|dS)a Create a remote HDFS directory. Args: fs_path(str): The HDFS directory path. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> client.mkdirs("hdfs:/test_hdfs_client") N)rrmkdirrr r r r"s zAFSClient.mkdirsTcCs^|r ||r |||r&||st|d||r&t|d|j||dS)a Move a remote HDFS file or directory from `fs_src_path` to `fs_dst_path` . Args: fs_src_path(str): Name of the file or directory, that's needed to be moved. fs_dst_path(str): Name of the file or directory to which to move to. overwrite(bool): Whether to re-write `fs_dst_path` if that exists. Default is False. test_exists(bool): Check the existence of `fs_src_path` and `fs_dst_path` . When `test_exists` is set true, if `fs_src_path` doesn't exist or `fs_dst_path` exists, program will throw an Excetption. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> client.mv("hdfs:/test_hdfs_client", "hdfs:/test_hdfs_client2") rrN)rr#rrrr-r*r r r r-s   z AFSClient.mvcCs||sdS|j|dS)a Delete a remote HDFS path, whether it's a file or directory. Args: fs_path(str): The HDFS file path. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> client.delete("hdfs:/test_hdfs_client") N)rrrKrr r r r#s zAFSClient.deletecCs"||r |r dSt|j|S)a~ Create a remote HDFS file. Args: fs_path(str): The HDFS file path. exist_ok(bool): When `fs_path` exists, if `exist_ok` is set false, program will throw an Exception. Default is true. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> client.touch("hdfs:/test_hdfs_client") N)rrrZtouchzr2r r r r43s  zAFSClient.touchcCrMrr r$r r r r%OrNzAFSClient.need_upload_downloadNcCs||r |j|SdS)a Cat a remote HDFS file. Args: fs_path(str): The HDFS file path. Returns: file content Examples: .. code-block:: python >>> # doctest: +SKIP('depend on WITH_PSLIB') >>> from paddle.distributed.fleet.utils.fs import AFSClient >>> client = AFSClient() >>> client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") >>> client.cat("hdfs:/test_hdfs_client") r)rrr5rr r r r5Rs  z AFSClient.catc Crrrrr r r rmrzAFSClient._split_filesrr)rFrr7r)rr r rYrrr1rrrrrrr0r r!r"r-r#r4r%r5rr r r r rs*    6 "   rr)r8rkrr<ryrGrr^Z paddle.baserZlog_utilr__all__rrrrrrrr:rorprr r r r s:  > )'