o p i@sdddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z m Z m Z mZmZddlmZddlmZe r`ddlmZedZed Zed eed ZGd d d eZgZGdddeZGdddeZGdddeZGdddeZ GdddeZ!GdddZ"Gddde"Z# d&d'd d!Z$Gd"d#d#e"Z%Gd$d%d%e"Z&dS)() annotationsN) TYPE_CHECKINGCallableLiteral TypedDictTypeVar)core)logger) ParamSpec_InputT_RetT_HDFSClientConfig)zfs.default.namezhadoop.job.ugic@seZdZUded<ded<dS) _FileInfostrpathintsizeN)__name__ __module__ __qualname____annotations__rrh/home/app/PaddleOCR-VL/.venv_paddleocr/lib/python3.10/site-packages/paddle/distributed/fleet/utils/fs.pyr,s  rc@ eZdZdS) ExecuteErrorNrrrrrrrr4rc@r)FSFileExistsErrorNrrrrrr8rrc@r)FSFileNotExistsErrorNrrrrrr<rrc@r) FSTimeOutNrrrrrr @rr c@r)FSShellCmdAbortedNrrrrrr!Drr!c@seZdZejddZejddZejddZejddZejd d Z ejd d Z ejd dZ ejddZ ejddZ ejddZejd"ddZejddZejddZejd#ddZejd$d d!ZdS)%FScCtNNotImplementedErrorselffs_pathrrrls_dirIz FS.ls_dircCr#r$r%r'rrris_fileMr+z FS.is_filecCr#r$r%r'rrris_dirQr+z FS.is_dircCr#r$r%r'rrris_existUr+z FS.is_existcCr#r$r%)r( local_pathr)rrruploadYr+z FS.uploadcCr#r$r%)r(r)r/rrrdownload]r+z FS.downloadcCr#r$r%r'rrrmkdirsar+z FS.mkdirscCr#r$r%r'rrrdeleteer+z FS.deletecCr#r$r%r(rrrneed_upload_downloadir+zFS.need_upload_downloadcCr#r$r%r( fs_src_path fs_dst_pathrrrrenamemr+z FS.renameFcCr#r$r%r(r7r8 overwrite test_existsrrrmvqr+zFS.mvcCr#r$r%)r( local_dirdest_dirrrr upload_dirur+z FS.upload_dircCr#r$r%r'rrr list_dirsyr+z FS.list_dirsTcCr#r$r%r(r)exist_okrrrtouch}r+zFS.touchNcCr#r$r%r'rrrcatr+zFS.catFFTr$)rrrabcabstractmethodr*r,r-r.r0r1r2r3r5r9r=r@rArDrErrrrr"Hs>              r"c@seZdZdZd.ddZd/d d Zd0d dZddZddZd/ddZ d1ddZ d2ddZ d2ddZ d2ddZ d3d4d!d"Z # #d5d6d(d)Zd7d+d,Zd-S)8LocalFSa( 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("./") r)rreturntuple[list[str], list[str]]cCs\||s ggfSg}g}t|D]}tj|d|r$||q||q||fS)a{ List directories 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("./") /)r.oslistdirrisdirappend)r(r)dirsfilesfrrrr*s   zLocalFS.ls_dirNonecCs,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)rCN)rNrisfilemakedirsr'rrrr2szLocalFS.mkdirsr7r8cCst||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)rNr9r6rrrr9szLocalFS.renamecCt|dSr$)shutilrmtreer'rrr_rmrz LocalFS._rmrcCrXr$)rNremover'rrr_rmr\z 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)r.rNrrVr^r[r'rrrr3s    zLocalFS.deleteLiteral[False]cCdS)NFrr4rrrr5zLocalFS.need_upload_downloadboolcC 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") )rNrrVr'rrrr,  zLocalFS.is_filecCrc)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") rNrrPr'rrrr-$rdzLocalFS.is_dircCrc)a Whether the local file path exists. Args: fs_path(str): The local file path. Returns: Bool: Whether 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") )rNrexistsr'rrrr.=s zLocalFS.is_existTrCcCsF||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)r.ropenrBrrrrDTs  "z LocalFS.touchFsrc_pathdst_pathr;r<cCs@||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") )r.rr3rr9)r(rirjr;r<rrrr=os    z LocalFS.mv list[str]cs*|sgSfddtD}|S)a Only list directories 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|qSrMre).0rTr)rr s z%LocalFS.list_dirs..)r.rNrO)r(r)rRrrnrrAs  zLocalFS.list_dirsNr)rrKrLr)rrKrU)r7rr8rrKrU)rKr_r)rrKrbrGr)rrCrbrKrUrF) rirrjrr;rbr<rbrKrUr)rrKrk)rrr__doc__r*r2r9r[r^r3r5r,r-r.rDr=rArrrrrJs"  "        &rJ max_time_out float | NonerK>Callable[[Callable[_InputT, _RetT]], Callable[_InputT, _RetT]]csdfdd }|S)NrTCallable[_InputT, _RetT]rKcstd fdd }|S) Nargs _InputT.argskwargs_InputT.kwargsrKr c s|d}}|durt|jd}n|d}t|jd}t}|} z|i|WStyY}z!t||krJtd|dt|t|WYd}~nd}~wwt|dkrttd|dt|t}q$)Nrg@@Tzargs:z timeout:zhadoop operator timeout:args:)float _time_out _sleep_intertimerr sleepprint)rzr|otime_outinterstartZlast_print_timee)rTrvrrhandlers2z2_handle_errors..decorator..handler)rzr{r|r}rKr ) functoolswraps)rTrrv)rTr decoratorsz!_handle_errors..decorator)rTryrKryr)rvrrrr_handle_errorss!rc@seZdZUdZded<  d]d^ddZd_ddZd_ddZed`ddZ edaddZ ddZ dd Z edbd"d#Z d$d%Zdbd&d'Zedbd(d)Z dcddd-d.Z  dedfd1d2Zed3d4Z  dedgd5d6Zed7d8Zedhd9d:Z  ;didjd?d@ZedAdBZdCdDZdEdFZedhdGdHZdkdldJdKZedLdMZdmdOdPZdndodSdTZedUdVZ dWdXZ!dpd[d\Z"dQS)q 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") ([], []) rk pre_commands hadoop_homerconfigsrrr sleep_interrKrUc Csg|_|d}|j|d}|j||r/|D]\}}d|d|} |j| q||_||_d|j|_td|_ dS)Nz /bin/hadoopfsz-D= z8\s?responseErrorMsg\s?\:.*, errorCode\:\s?[0-9]+, path\:) rrQitemsrrjoin _base_cmdrecompile _bd_err_re) r(rrrrZ hadoop_bindfskvZconfig_commandrrr__init__s    zHDFSClient.__init__Fc Cs||jd|}d}d}d}t|dD]}t|dd|\}}t|}|dkr*nt|q|dkr8t|||fS)Nz -rr ) rrangerZshell_execute_cmdrrrr! splitlines) r(cmdredirect_stderr retry_timesexe_cmdretoutputretry_sleep_secondxrrr_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)Nrrr T)checkstdoutstderrtextr)rsplitr subprocessrunPIPESTDOUTrCalledProcessError returncoderrr Exceptionr!) r(rrrrrrrrprocessrrrr _run_safe_cmd!s: zHDFSClient._run_safe_cmdr)cC ||sgS||\}}|S)a* Only list directories 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_dirr(r)rRrSrrrrA?s zHDFSClient.list_dirsrLcC||s ggfS||S)a List directories 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") rr'rrrr*as  zHDFSClient.ls_dirc Csd|}||\}}|dkrt|g}g}|D](}|}t|dkr'qtj|d} |dddkr=|| q|| q||fS)Nls rd)rrrlenrNrbasenamerQ) r(r)rrlinesrRrSlinearrprrrrs    zHDFSClient._ls_dircCs*|D]}|j|}|dur|SqdSr$)rmatch)r(rlmrrr _test_matchs  zHDFSClient._test_matchrbcC||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_dirr'rrrr-s  zHDFSClient.is_dircCsRd|}|j|ddd\}}|r'||r%tdtd|t|dSdS)Nztest -d Tr rrzraise exception:  F)rrrrr)r(r)rrrrrrrs  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") Frr'rrrr,s  zHDFSClient.is_filecCs0d|d}|j|ddd\}}|dkrdSdS)aA 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: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_exist("hdfs:/test_hdfs_client") ztest -e rTr rrF)r)r(r)rroutrrrr.s zHDFSClient.is_existr>r?r;cCsl|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 rMN)rstriprNrrr.r3r2 _try_uploadr(r>r?r;Zlocal_basenamerrrr@s    zHDFSClient.upload_dirr/multi_processescsfdd}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_singledatasdatar4rrZ__subprocess_uploadLsz.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 )rNrrfrPrOrrQ)rZrlistfiletrrrget_local_filesPs    z*HDFSClient.upload..get_local_files not existsz/there are nothing need to upload, function exitNtargetrz) rJr.rrr3r2r _split_filesmultiprocessingProcessrQrr)r(r/r)rr;Z_HDFSClient__subprocess_uploadrlocal all_filesprocsi process_datasrprocrr4rr0*s. "       zHDFSClient.uploadc Cs`d|d|}d}z||\}}|dkrt|WdSty/}z|||d}~ww)Nzput rr)rrrr3)r(r/r)rr_rrrrrs 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]}||qdS)z download file from HDFS Args: local_path(str): the local file path datas(str): the hdfs file path list N) _try_download)r/rrr4rrZ__subprocess_downloadsz2HDFSClient.download..__subprocess_download not exitscg|]}d|qSrlrrmrrnrrroz'HDFSClient.download..crrlrrrnrrrorrN) r.rr,rr*extendrrrrrQrr)r(r)r/rr;Z _HDFSClient__subprocess_downloadrR all_filenamesrrrrrrr)r)r(rr1s& "       zHDFSClient.downloadc Csfd|d|}d}z||\}}|dkrt|WdSty2}z t}|||d}~ww)Nzget rr)rrrrJr3)r(r)r/rrrrZlocal_fsrrrrs 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 rT)rrzNo such file or directoryz mkdir -p )r.rr)r(r)Zout_hdfsrrrrrrrrr2s(   zHDFSClient.mkdirsTr7r8r<cCsX|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 Exception. 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)r.r3rr_try_mvr:rrrr= s!    z HDFSClient.mvc Cs|d|d|}d}z|j|dd\}}|dkrt|WdSty=}z||s7||r7WYd}~dS|d}~ww)Nzmv rrr r)rrrr.)r(r7r8rrrrrrrr6szHDFSClient._try_mvcC,d|}||\}}|dkrt|dS)Nzrmr rrrr(r)rrrrrrr[C zHDFSClient._rmrcCr)Nzrm rrrrrrr^IrzHDFSClient._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)r.rr[r^)r(r)r-rrrr3Os    zHDFSClient.deleterCcCs ||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)r.r_touchzrBrrrrDqs  zHDFSClient.touchcCr)Nztouchz rrrrrrrs zHDFSClient._touchz Literal[True]cCr`NTrr4rrrr5razHDFSClient.need_upload_downloadN str | NonecCs"||r||}d|SdS)a Cat a remote HDFS file. Args: fs_path(str|None): 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_catr)r(r)rrrrrEs   zHDFSClient.catcCs0d|}|j|dd\}}|dkrt||S)Nzcat r rrr)r(r)rrrrrrrs 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: filelist(list): file list of current trainer r rrr r(rSZ trainer_idZtrainers remainder blocksizeblocksrZ trainer_filesbeginrrrr     zHDFSClient._split_files path_listlist[_FileInfo]c Cst|dkrgSg}d}|D]}||d7}qd|d}||\}}t|dkr5td|dgS|D] }|d} t| dkrEq7| d } t| d} || | d q7|S) z list_files return file path and size Args: path_list(list): file list Returns: filelist(list): file list with file path and size rrrrz) | awk '{if ($8 != "") {print $5" "$8 }}'zlist_files empty, path[]r )rr)rrr warningrrrQ) r(r file_listZ str_concatrrrrrr file_path file_sizerrrlist_files_infos(      zHDFSClient.list_files_inforr) rrrrrrrrrKrU)FrrtrprrF)r>rr?rr;rbrKrU)rF) r/rr)rrrr;rbrKrU) r)rr/rrrr;rbrKrUrqFT) r7rr8rr;rbr<rbrKrUrGrs)rKrr$)r)rrKr)rrkrKr)#rrrrurrrrrrAr*rrr-rr,r.r@r0rr1rr2r=rr[r^r3rDrr5rErrrrrrrrsh    ! !   $  U  B  1 -    !!   " rc@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. When WITH_PSLIB=ON, you can use this class directly. When WITH_PSCORE=ON, you should export LD_LIBRARY_PATH='YOUR_AFSAPISO_PATH' before using this class. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on external file') >>> 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") rrcCst|_||_dSr$)rZ AfsWrapper_fsr)r(rrrrrrs  zAFSClient.__init__cCs|j||||dSr$)rinit)r(Zfs_nameZfs_userZ fs_passwdZfs_confrrrrszAFSClient.initcCr)a{ Only list directories 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 external file') >>> 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") rrrrrrAs zAFSClient.list_dirscCr)a  List directories 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 external file') >>> 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") rr'rrrr*8s  zAFSClient.ls_dircCs|j|}|g}||fSr$)rlist)r(r)rSrRrrrrTs 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 external file') >>> 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") Frr'rrrr-Ys  zAFSClient.is_dircCs |j|}t|dkrdSdS)NrTF)rrr)r(r) list_pathrrrrts  zAFSClient._is_dircCr)au 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 external file') >>> 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") Frr'rrrr,{s  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 external file') >>> 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") )rexistr'rrrr.rdzAFSClient.is_existFcCsn|d}|d}tj|}||d|r$|r$||d|||s.|||j||dSr) rrNrrr.r3r2rr0rrrrr@s    zAFSClient.upload_dirr cCs0t}||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 external file') >>> 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)rJr.rrr0)r(r/r)rr;rrrrr0s zAFSClient.uploadc Cst||s t|d||r|j||S||\}}|D]}tj|tj |d}|j||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:: python >>> # doctest: +SKIP('depend on external file') >>> 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", "./") rr N) r.rr,rr1r*rNrrr) r(r)r/rr;rr file_nameZlocal_file_namerrrr1s  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 external file') >>> 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)r.rmkdirr'rrrr2s 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 Exception. Examples: .. code-block:: python >>> # doctest: +SKIP('depend on external file') >>> 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)r.r3rrrr=r:rrrr=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 external file') >>> 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)r.rr]r'rrrr3>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 external file') >>> 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)r.rrZtouchzrBrrrrDVs  zAFSClient.touchcCr`rrr4rrrr5rrazAFSClient.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 external file') >>> 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)r,rrEr'rrrrEus  z AFSClient.catc Crrrr rrrrrzAFSClient._split_filesrr)r FrrGr$)rrrrurrrAr*rr-rr,r.r@r0r1r2r=r3rDr5rErrrrrrs*    % "   rr$)rvrwrKrx)' __future__rrHrrrNrrYrrtypingrrrrrZ paddle.baserZlog_utilr Ztyping_extensionsr r r rrr__all__rrrrr r!r"rJrrrrrrrsP     >0 '.