o Meƻ@s.ddlZddlZddlZddlZddlmZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl m Z ddl mZddlZddlZgZGdddeZGdddeZGdd d eZGd d d eZGd d d eZGdddeZGdddeZdddZGdddeZGdddeZdS)N)datetime)corec@ eZdZdS) ExecuteErrorN__name__ __module__ __qualname__r r QD:\Projects\ConvertPro\env\Lib\site-packages\paddle/distributed/fleet/utils/fs.pyr%rc@r)FSFileExistsErrorNrr r r r r )r r c@r)FSFileNotExistsErrorNrr r r r r-r rc@r) FSTimeOutNrr r r r r1r rc@r)FSShellCmdAbortedNrr r r r r5r rc@seZdZejddZejddZejddZejddZejd d Z ejd d Z ejd dZ ejddZ ejddZ ejddZejd"ddZejddZejddZejd#ddZejd$d d!ZdS)%FScCtNNotImplementedErrorselffs_pathr r r ls_dir;z FS.ls_dircCrrrrr r r is_file?rz FS.is_filecCrrrrr r r is_dirCrz FS.is_dircCrrrrr r r is_existGrz FS.is_existcCrrr)r local_pathrr r r uploadKrz FS.uploadcCrrr)rrrr r r downloadOrz FS.downloadcCrrrrr r r mkdirsSrz FS.mkdirscCrrrrr r r deleteWrz FS.deletecCrrrrr r r need_upload_download[rzFS.need_upload_downloadcCrrrr fs_src_path fs_dst_pathr r r rename_rz FS.renameFcCrrrrr&r' overwrite test_existsr r r mvcrzFS.mvcCrrr)r local_dirdest_dirr r r upload_dirgrz FS.upload_dircCrrrrr r r list_dirskrz FS.list_dirsTcCrrrrrexist_okr r r touchorzFS.touchNcCrrrrr r r catsrzFS.catFFTr)rrr abcabstractmethodrrrrrr r!r"r$r(r,r/r0r3r4r r r r r9s>              rc@s|eZdZdZddZddZddZdd Zd d Zd d Z ddZ ddZ ddZ ddZ dddZd ddZddZdS)!LocalFSz A tool of local file system. Examples: .. code-block:: python 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)a5 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 from paddle.distributed.fleet.utils import LocalFS client = LocalFS() subdirs, files = client.ls_dir("./") /)roslistdirpathisdirappend)rrdirsfilesfr r r rs   zLocalFS.ls_dircCs.tj|r Jd|td|dS)af Create a local directory. Args: fs_path(str): The local directory path. Examples: .. code-block:: python from paddle.distributed.fleet.utils import LocalFS client = LocalFS() client.mkdirs("test_mkdirs") client.delete("test_mkdirs") z{} is already a file mkdir -p {}N)r;r=isfileformatsystemrr r r r!szLocalFS.mkdirscCst||dS)a 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 from paddle.distributed.fleet.utils import LocalFS client = LocalFS() client.touch("test_rename_src") print(client.is_exists("test_rename_src")) # True client.rename("test_rename_src", "test_rename_dst") print(client.is_exists("test_rename_src")) # False print(client.is_exists("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._rmrcCrGr)r;removerr r r _rmrKz 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 from paddle.distributed.fleet.utils import LocalFS client = LocalFS() client.mkdirs("test_localFS_mkdirs") client.delete("test_localFS_mkdirs") N)rr;r=rDrMrJrr r r r"s    zLocalFS.deletecCdS)NFr r#r r r r$zLocalFS.need_upload_downloadcC tj|S)a 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 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_filecCrP)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 from paddle.distributed.fleet.utils import LocalFS client = LocalFS() client.mkdirs("test_is_dir") print(client.is_dir("test_is_file")) # True client.delete("test_is_dir") r;r=r>rr r r rrQzLocalFS.is_dircCrP)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 from paddle.distributed.fleet.utils import LocalFS client = LocalFS() ret = local_fs.is_exist("test_is_exist") )r;r=existsrr r r rs zLocalFS.is_existTcCs*||r |r dSttd|dS)a 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 from paddle.distributed.fleet.utils import LocalFS client = LocalFS() client.touch("test_touch") client.delete("test_touch") Nztouch {})rr r;rFrEr1r r r r32s 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 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"r r()rZsrc_pathZdst_pathr*r+r r r r,Ks    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 from paddle.distributed.fleet.utils import LocalFS client = LocalFS() subdirs = client.list_dirs("./") cs$g|]}tjd|r|qSr:rR).0rBrr r ~s z%LocalFS.list_dirs..)rr;r<)rrr@r rVr r0is  zLocalFS.list_dirsNr6r5)rrr __doc__rr!r(rJrMr"r$rrrr3r,r0r r r r r9xs     r9csfdd}|S)Ncstfdd}|S)Nc s|d}}|durt|jd}n|d}t|jd}t}|} z|i|WStyW}zt||krHtd|t|t|WYd}~nd}~wwt|dkrptd|t|t}q$)Nrg@@Tzargs:{} timeout:{}z*hadoop operator timeout:args:{} timeout:{}) float _time_out _sleep_intertimerrrEsleepprint)argskwargsotime_outinterstartZlast_print_timee)rB max_time_outr r handlers6  z2_handle_errors..decorator..handler) functoolswraps)rBrhrg)rBr decoratorsz!_handle_errors..decoratorr )rgrlr rkr _handle_errorss "rmc@sFeZdZdZ  d@ddZdAdd Zed d Zed d ZddZ ddZ eddZ ddZ ddZ eddZdBddZdCddZeddZdCd d!Zed"d#Zed$d%ZdDd'd(Zed)d*Zd+d,Zd-d.Zed/d0ZdEd1d2Zed3d4Zd5d6ZdFd8d9Zed:d;Zdd?Z d7S)G 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:: text 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-t|D]\}}d||f} |j| q||_||_d|j|_t d|_ dS)Nz %s/bin/hadoopfsz-D%s=%s z8\s?responseErrorMsg\s?\:.*, errorCode\:\s?[0-9]+, path\:) Z pre_commandsr?six iteritemsr[r\join _base_cmdrecompile _bd_err_re) rZ hadoop_homeZconfigsrc sleep_interZ hadoop_bindfskvZconfig_commandr r r __init__s    zHDFSClient.__init__Fc Cszd|j|}d}d}d}t|dD]}t|dd|\}}t|}|dkr)nt|q|dkr7t||| fS)Nz{} -{}r) rErvrangerZshell_execute_cmdintr]r^r splitlines) rcmdredirect_stderr retry_timesZexe_cmdretoutputZretry_sleep_secondxr r r _run_cmds  zHDFSClient._run_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:: text 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_dirrrr@rAr r r r0s 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:: text 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 r s  zHDFSClient.ls_dirc Csd|}||\}}|dkrt|g}g}|D](}|}t|dkr'qtj|d} |dddkr=|| q|| q||fS)Nzls {}rd) rErrsplitlenr;r=basenamer?) rrrrlinesr@rAlinearrpr r r r,s    zHDFSClient._ls_dircCs*|D]}|j|}|dkr|SqdSr)rymatch)rrlmr r r _test_matchBs  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:: text 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 rJs  zHDFSClient.is_dircCsTdj|dd}|j|dd\}}|r(||r&tdtd|t|dSdS) Nz test -d {}Trrrzraise exception:  F)rErrr_rur)rrrrrr r r ris 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:: text 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 rws  zHDFSClient.is_filecCs.d|}|j|ddd\}}|dkrdSdS)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:: text 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") z test -e {} Tr)rrrF)rEr)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 r/s    zHDFSClient.upload_dircsfdd}dd}t}||std|||}|s%tddS|r6|r6||g} t|D]} || |} t j ||| 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:: text 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_uploadsz.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=rSr>r<rur?)r=Zrlistfiletr r r get_local_filess    z*HDFSClient.upload..get_local_files {} not existsz/there are nothing need to upload, function exitNtargetr`)r9rrrEr_r"r!r _split_filesmultiprocessingProcessr?reru)rrrmulti_processesr*Z_HDFSClient__subprocess_uploadrlocal all_filesprocsi process_datasrprocr r#r rs.        zHDFSClient.uploadc Cs\d||}d}z||\}}|dkrt|WdSty-}z|||d}~ww)Nz put {} {}r)rErr Exceptionr")rrrrr_rfr r r rs  zHDFSClient._try_uploadcsfdd}stdr|S\}}fdd|D}|fdd|Dg} t|D]} || |} t j ||| fd} | | | q?| D]} | q]dS) 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:: text 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_download>sz2HDFSClient.download..__subprocess_download {} not exitscg|]}d|qSrTr rUrrVr r rWOz'HDFSClient.download..crrTr rrVr r rWPrrN)rrrErrrextendrrrrr?reru)rrrrr*Z _HDFSClient__subprocess_downloadr@ all_filenamesrrrrrrr rrr r $s&        zHDFSClient.downloadc Csbd||}d}z||\}}|dkrt|WdSty0}z t}|||d}~ww)Nz get {} {}r)rErrrr9r")rrrrrrrfZlocal_fsr r r r]s  zHDFSClient._try_downloadcCs||rdSd}d|}|j|dd\}}|dkr.|D] }d|vr'd}nq|s.t||rI||sKd|}||\}}|dkrMt|dSdSdS) aQ Create a remote HDFS directory. Args: fs_path(str): The HDFS directory path. Examples: .. code-block:: text 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") NFz mkdir {} TrrzNo such file or directoryrC)rrErr)rrZout_hdfsrrrrrr r r r!js(   zHDFSClient.mkdirsTcCsX|r ||r |||r&||std|||r&td||||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:: text 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"rrEr _try_mvr)r r r r,s    z HDFSClient.mvc Csxd||}d}z|j|dd\}}|dkrt|WdSty;}z||s5||r5WYd}~dS|d}~ww)Nzmv {} {}rrr)rErrrr)rr&r'rrrrfr r r rs  zHDFSClient._try_mvcC,d|}||\}}|dkrt|dS)Nzrmr {}rrErrrrrrrr r r rJ zHDFSClient._rmrcCr)Nzrm {}rrrr r r rMrzHDFSClient._rmcCs0||sdS||}|r||S||S)ai Delete a remote HDFS path, whether it's a file or directory. Args: fs_path(str): The HDFS file path. Examples: .. code-block:: text 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)rrrJrM)rrrr r r r"s    zHDFSClient.deletecCs ||r |r dSt||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:: text 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 _touchzr1r r r r3s  zHDFSClient.touchcCr)Nz touchz {}rrrr r r rs zHDFSClient._touchzcCrNNTr r#r r r r$rOzHDFSClient.need_upload_downloadNcCs"||r||}d|SdS)al Cat a remote HDFS file. Args: fs_path(str): The HDFS file path. Returns: file content Examples: .. code-block:: text 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") r)r_try_catru)rrrr r r r4s   zHDFSClient.catcCs0d|}|j|dd\}}|dkrt||S)Nzcat {}rrrr)rrrrrr r r r>s 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 rrAZ trainer_idZtrainers remainder blocksizeblocksrZ trainer_filesbeginr r r rF     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 rrrrzls z) | awk '{if ($8 != "") {print $5" "$8 }}'zlist_files empty, path[%s]r)r=size)rrloggerwarningrrr?) r path_list file_listZ str_concatr=rrrrr file_path file_sizer r r list_files_info_s&      zHDFSClient.list_files_infororp)FrF)rFFTr6r)!rrr rXr~rrmr0rrrrrrrr/rrr rr!r,rrJrMr"r3rr$r4rrrr r r r rnsV      ! L  9   +&        rnc@seZdZdZ  d.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 AFSClientaE A tool of AFS. Use AfsWrapper. Examples: .. code-block:: text from paddle.distributed.fleet.utils import AFSClient client = AFSClient() client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") client.ls_dir("hdfs:/test_hdfs_client") rorpcCst|_||_dSr)rZ AfsWrapper_fsr[)rrcrzr r r r~s  zAFSClient.__init__cCs|j||||dSr)rinit)rZfs_nameZfs_userZ fs_passwdZfs_confr r r rszAFSClient.initcCr)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:: text from paddle.distributed.fleet.utils 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 r0s 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:: text from paddle.distributed.fleet.utils 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)rrrAr@r r r rs 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:: text from paddle.distributed.fleet.utils 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 rs  zAFSClient.is_dircCs |j|}t|dkrdSdS)NrTF)rrr)rr list_pathr r r rs  zAFSClient._is_dircCr)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:: text from paddle.distributed.fleet.utils 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 rs  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:: text from paddle.distributed.fleet.utils 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 r s zAFSClient.is_existFcCsn|d}|d}tj|}||d|r$|r$||d|||s.|||j||dSr) rr;r=rrr"r!rrrr r r r/"s    zAFSClient.upload_dirrcCs0t}||std||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:: text from paddle.distributed.fleet.utils 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)r9rrrErr)rrrrr*rr r r r6s zAFSClient.uploadcsfdd}stdrj|S\}}fdd|D}g} t|D]} || |} t j ||| fd} | | | q4| D]} | qRdS)aa 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:: text from paddle.distributed.fleet.utils 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 rfsz1AFSClient.download..__subprocess_downloadrcsg|]}|qSr r rrVr r rWwsz&AFSClient.download..rN)rrrErrr rrrrrr?reru)rrrrr*Z_AFSClient__subprocess_downloadrrrrrrrrr rr r Qs$       zAFSClient.downloadcCs||rdS|j|dS)a Create a remote HDFS directory. Args: fs_path(str): The HDFS directory path. Examples: .. code-block:: text from paddle.distributed.fleet.utils 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&||std|||r&td||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:: text from paddle.distributed.fleet.utils 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"rrEr rr,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:: text from paddle.distributed.fleet.utils import HDFSClient from paddle.distributed.fleet.utils import AFSClient client = AFSClient() client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") client.delete("hdfs:/test_hdfs_client") N)rrrLrr 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:: text from paddle.distributed.fleet.utils import AFSClient client = AFSClient() client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") client.touch("hdfs:/test_hdfs_client") N)rr rZtouchzr1r r r r3s  zAFSClient.touchcCrNrr r#r r r r$rOzAFSClient.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:: text from paddle.distributed.fleet.utils import AFSClient client = AFSClient() client.init("hdfs://xxx.hadoop.com:54310", "hello", "hello123", "./fs_conf") client.cat("hdfs:/test_hdfs_client") r)rrr4rr r r r4s  z AFSClient.catc Crrrrr r r rrzAFSClient._split_filesrr)rFrr6r)rrr rXr~rr0rrrrrrr/rr r!r,r"r3r$r4rr r r r rs.    3 !   rr)r;sys subprocessrrrwcopyerrnor]loggingrsr7Z paddle.fluidZfluidrrirH__all__rrr rrrobjectrr9rmrnrr r r r sD   ? 'Y