x-jddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z m Z m Z mZmZddlmZddlmZe r9ddlmZedZed Zed eed ZGd d eZgZGddeZGddeZGddeZGddeZ GddeZ!GddZ"Gdde"Z# d%d&d Z$Gd!d"e"Z%Gd#d$e"Z&dS)') annotationsN) TYPE_CHECKINGCallableLiteral TypedDictTypeVar)core)logger) ParamSpec_InputT_RetT_HDFSClientConfig)zfs.default.namezhadoop.job.ugic$eZdZUded<ded<dS) _FileInfostrpathintsizeN)__name__ __module__ __qualname____annotations__a/var/www/html/banglarbhumi/venv/lib/python3.11/site-packages/paddle/distributed/fleet/utils/fs.pyrr,s"  rrceZdZdS) ExecuteErrorNrrrrrrrr4DrrceZdZdS)FSFileExistsErrorNrrrrr"r"8r rr"ceZdZdS)FSFileNotExistsErrorNrrrrr$r$<r rr$ceZdZdS) FSTimeOutNrrrrr&r&@r rr&ceZdZdS)FSShellCmdAbortedNrrrrr(r(Dr rr(ceZdZejdZejdZejdZejdZejdZ ejdZ ejdZ ejdZ ejd Z ejd Zejdd Zejd ZejdZejddZejddZdS)FSctNNotImplementedErrorselffs_paths rls_dirz FS.ls_dirI!!rctr,r-r/s ris_filez FS.is_fileMr3rctr,r-r/s ris_dirz FS.is_dirQr3rctr,r-r/s ris_existz FS.is_existUr3rctr,r-)r0 local_pathr1s ruploadz FS.uploadYr3rctr,r-)r0r1r;s rdownloadz FS.download]r3rctr,r-r/s rmkdirsz FS.mkdirsar3rctr,r-r/s rdeletez FS.deleteer3rctr,r-r0s rneed_upload_downloadzFS.need_upload_downloadir3rctr,r-r0 fs_src_path fs_dst_paths rrenamez FS.renamemr3rFctr,r-r0rHrI overwrite test_existss rmvzFS.mvqr3rctr,r-)r0 local_dirdest_dirs r upload_dirz FS.upload_dirur3rctr,r-r/s r list_dirsz FS.list_dirsyr3rTctr,r-r0r1exist_oks rtouchzFS.touch}r3rNctr,r-r/s rcatzFS.catr3rFFTr,)rrrabcabstractmethodr2r5r7r9r<r>r@rBrErJrOrSrUrYr[rrrr*r*Hs""" """ """ """ """ """ """ """ """ """ """" """ """ """" """"""rr*c~eZdZdZd!dZd"dZd#d Zd Zd Zd"dZ d$dZ d%dZ d%dZ d%dZ d&d'dZ d(d)dZd*dZd S)+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("./") r1rreturntuple[list[str], list[str]]c||sggfSg}g}tj|D]R}tj|dz|zr||=||S||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("./") /)r9oslistdirrisdirappend)r0r1dirsfilesfs rr2zLocalFS.ls_dirs*}}W%% r6MG$$  Aw}}Ws]Q.//  A QU{rNonectj|r J|dtj|ddS)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)rXN)rfrisfilemakedirsr/s rr@zLocalFS.mkdirssJ$7>>'**JJw,J,J,JJJ* Gd++++++rrHrIc0tj||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)rfrJrGs rrJzLocalFS.renames4 +{+++++rc.tj|dSr,)shutilrmtreer/s r_rmrz LocalFS._rmrs grc.tj|dSr,)rfremover/s r_rmz LocalFS._rms 'rc||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)r9rfrrorxrur/s rrBzLocalFS.deletesW$}}W%%  F 7>>' " " %88G$$ $yy!!!rLiteral[False]cdS)NFrrDs rrEzLocalFS.need_upload_downloadsurboolc@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") )rfrror/s rr5zLocalFS.is_file s.w~~g&&&rc@tj|S)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") rfrrhr/s rr7zLocalFS.is_dir$s.w}}W%%%rc@tj|S)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") )rfrexistsr/s rr9zLocalFS.is_exist=s*w~~g&&&rTrXc||r |rdStt|d5 ddddS#1swxYwYdS)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)r9r"openrWs rrYz LocalFS.touchTs( == ! ! $ # # '3                       sAAAFsrc_pathdst_pathrMrNc||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") )r9r$rBr"rJ)r0rrrMrNs rrOz LocalFS.mvosy6}}X&& '& &  "x00 " KK ! ! ! == " " $# #{{8X...r list[str]cv|sgSfdtjD}|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("./") c^g|])}tjdz|z'|*Srer).0rlr1s r z%LocalFS.list_dirs..sE   bgmmGcMA ! $/$/$/$/$/Lrra max_time_out float | Nonerb>Callable[[Callable[_InputT, _RetT]], Callable[_InputT, _RetT]]cdfd }|S)NrlCallable[_InputT, _RetT]rbcLtjdfd }|S) Nargs _InputT.argskwargs_InputT.kwargsrbrcL|d} }|t|jdz }n|dz}t|jdz }tj}|} |i|S#t$ra}tj|z |kr)t d|dtj|z tj|Yd}~nd}~wwxYwtj|z dkr.decorator..handlers[QA#H --6F"!.))F2EIKKE#O 2 &1d-f---#&&&y{{U*h66'HDHH49;;3FHHJu%%%%%%%% &9;;0255\\\ty{{UZGZ\\'+ikkO! 2sA!! C +ACC )rrrrrbr) functoolswraps)rlrrs` r decoratorz!_handle_errors..decoratorsC    2 2 2 2 2 2   2:r)rlrrbrr)rrs` r_handle_errorsrs*B rcpeZdZUdZded< d@dAdZdBdZdBdZedCdZ edDdZ dZ dZ edEdZ dZdEdZedEdZ dFdGd"Z dHdId%Zed&Z dHdJd'Zed(ZedKd)Z dLdMd.Zed/Zd0Zd1ZedKd2ZdNdOd4Zed5ZdPd7ZdQdRd:Zed;Z d<Z!dSd?Z"d8S)T 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") ([], []) r pre_commands hadoop_homerconfigsrrr sleep_interrbrmcg|_|d}|j|d}|j||r<|D]'\}}d|d|} |j| (||_||_d|j|_tjd|_ dS)Nz /bin/hadoopfsz-D= z8\s?responseErrorMsg\s?\:.*, errorCode\:\s?[0-9]+, path\:) rriitemsrrjoin _base_cmdrecompile _bd_err_re) r0rrrr hadoop_bindfskvconfig_commands r__init__zHDFSClient.__init__s#000    ,,,   %%%  9  9 91!-a!!((8888!'$"344* G  rFc0|jd|}d}d}d}t|dzD]G}tj|dd|\}}t |}|dkrnt j|H|dkrt|||fS)Nz -rr ) rranger shell_execute_cmdrrrr( splitlines) r0cmdredirect_stderr retry_timesexe_cmdretoutputretry_sleep_secondxs r_run_cmdzHDFSClient._run_cmds^,,s,,{Q'' + +A0!QPPKCc((Caxx J) * * * * #::#C(( (F%%''''rc|jg|}d}d}d}t|dzD]} tj|dtj|r tjn tjd} | j}nQ#tj$r,} | j }| j }tj |Yd} ~ }d} ~ wt$r } Yd} ~ nd} ~ wwxYw|dkrt|dS)Nrrr T)checkstdoutstderrtextr)rsplitr subprocessrunPIPESTDOUTrCalledProcessError returncoderrr Exceptionr() r0rrrrrrrrprocessrs r _run_safe_cmdzHDFSClient._run_safe_cmd!s%>0CIIKK0{Q''  A $.%?+- ))'_   !0 / / /l -........     #::#C(( ( :sAA<<C  "B22 C C r1cd||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") r9_ls_dirr0r1rjrks rrUzHDFSClient.list_dirs?s78}}W%% Ill7++ e rrcc^||sggfS||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/s rr2zHDFSClient.ls_diras3:}}W%% r6M||G$$$rcd|}||\}}|dkrt|g}g}|D]}|}t|dkr*tj|d} |dddkr|| w|| ||fS)Nls rd)rrrlenrfrbasenameri) r0r1rrlinesrjrklinearrps rrzHDFSClient._ls_dirsGoo]]3'' U !88s## # D**,,C3xx1}}  Q((A1vayC A QU{rcP|D]"}|j|}||cS#dSr,)rmatch)r0rlms r _test_matchzHDFSClient._test_matchs@  A%%a((A}trr|cZ||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") Fr9_is_dirr/s rr7zHDFSClient.is_dirs/8}}W%% 5||G$$$rcd|}||dd\}}|rW||r@tdtd|t |dSdS)Nztest -d Tr rrzraise exception:  F)rrrrr)r0r1rrrs rrzHDFSClient._is_dirs"""]]3!]LL U  && ()***dii&&'''"3'''5trc\||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/s rr5zHDFSClient.is_files26}}W%% 5<<((((rcXd|d}||dd\}}|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)r0r1rrouts rr9zHDFSClient.is_existsB:$###==d=JJS !885trrQrRrMc|d}|d}tj|}||dz|zr|r||dz|z||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 reN)rstriprfrrr9rBr@ _try_uploadr0rQrRrMlocal_basenames rrSzHDFSClient.upload_dirs$$S)) ??3'')))44 ==C.8 9 9 9i 9 KK37 8 8 8}}X&& " KK ! ! ! H-----rr;multi_processescJfd}d}t}||st|d||}|stddS|r,|r*||g} t |D]Z} || |} tj ||| f} | | | [| D]} | 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 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") c>|D]}||dSr,)r )hdfs_path_singledatasdatar0s r__subprocess_uploadz.HDFSClient.upload..__subprocess_uploadLs6 9 9  '78888 9 9rcNg}tj|s|Stj|rMtj|D]7}tj||}||8n|||S)z get local files Args: path(str): local path Returns: list of local files )rfrrrhrgrri)rrlistfilets rget_local_filesz*HDFSClient.upload..get_local_filesPsE7>>$''  w}}T"" #Jt,,$$D T400ALLOOOO$ T"""Lr not existsz/there are nothing need to upload, function exitNtargetr) rar9r$rrBr@r _split_filesmultiprocessingProcessrirr)r0r;r1r rM_HDFSClient__subprocess_uploadrlocal all_filesprocsi process_datasrprocs` rr<zHDFSClient.upload*skD 9 9 9 9 9   * ~~j)) C&*'A'A'ABB B#OJ//   C D D D F == ! ! !i ! KK KK ''  A --iOLLM'*'=1IA LLOOO GGIIII  D IIKKKK  rcd|d|}d} ||\}}|dkrt|dS#t$r}|||d}~wwxYw)Nzput rr)rrrrB)r0r;r1rr_rs rr zHDFSClient._try_uploads+Z++'++ ]]3''FCaxx"3'''x    KK G s-; A!AA!cPfd}stdr|S\}}fd|D}|fd|Dg} t |D]Z} || |} tj ||| f} | | | [| D]} | 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", "./") c>|D]}||dS)z download file from HDFS Args: local_path(str): the local file path datas(str): the hdfs file path list N) _try_download)r;rrr0s r__subprocess_downloadz2HDFSClient.download..__subprocess_downloads7 5 5""44444 5 5r not exitsc g|] }dz|z Srrrr#r1s rrz'HDFSClient.download..s">>>1Ws]Q&>>>rc g|] }dz|z Srrr.s rrz'HDFSClient.download..s":::'C-!+:::rrN) r9r$r5r*r2extendrrrrrirr)r0r1r;r rM _HDFSClient__subprocess_downloadrj all_filenamesr!r"r#r$rr%s`` rr>zHDFSClient.downloadskD 5 5 5 5 5}}W%% ?&''='='=>> > <<  ;%%gz:: :"kk'22m>>>> >>> ::::T:::;;;''  A --iOLLM',J 3NA LLOOO GGIIII  D IIKKKK  rcd|d|}d} ||\}}|dkrt|dS#t$r*}t}|||d}~wwxYw)Nzget rr)rrrrarB)r0r1r;rrr'rlocal_fss rr*zHDFSClient._try_downloads+W++z++ ]]3''FCaxx"3'''x   yyH OOJ ' ' 'G s-; A/%A**A/cZ||rdSd}d|d}||d\}}|dkr|D] }d|vrd}n |st||rG||s4d |}||\}}|dkrt|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 )r9rr)r0r1out_hdfsrrrrr's rr@zHDFSClient.mkdirss2 == ! !  F!w!!!==d=;;S !88  .!33#HE4 ("3'''  (DMM'22 ('g''C]]3''FCaxx"3'''  ( ( ( (xrTrHrIrNc&|r*||r|||rN||st|d||rt|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)r9rBr$r"_try_mvrLs rrOz HDFSClient.mv sB  %{33 % KK $ $ $  I==-- K*k+I+I+IJJJ}}[)) I';(G(G(GHHH||K555rcd|d|}d} ||d\}}|dkrt|dS#t$r7}||s||rYd}~dS|d}~wwxYw)Nzmv rrr r)rrrr9)r0rHrIrrr'rs rr:zHDFSClient._try_mv6s/K//+// ]]3A]66FCaxx"3'''x   ==-- $-- 2L2L G s/= A>*A97A99A>cjd|}||\}}|dkrt|dS)Nzrmr rrrr0r1rrr's rruzHDFSClient._rmrCsBWs##Q !88s## # 8rcjd|}||\}}|dkrt|dS)Nzrm rr>r?s rrxzHDFSClient._rmIs@Goos##Q !88s## # 8rc||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)r9rrurx)r0r1r7s rrBzHDFSClient.deleteOsZ2}}W%%  Fg&&  &99W%% %xx   rrXcl||r |rdSt||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)r9r"_touchzrWs rrYzHDFSClient.touchqs=4 == ! ! $ # #||G$$$rcjd|}||\}}|dkrt|dS)Nztouchz rr>r?s rrCzHDFSClient._touchzsB!!!s##Q !88s## # 8r Literal[True]cdSNTrrDs rrEzHDFSClient.need_upload_downloadtrN str | Nonec||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)r5_try_catr)r0r1rs rr[zHDFSClient.cats@8 <<  ]]7++F99V$$ $2rcnd|}||d\}}|dkrt||S)Nzcat r r<rr>)r0r1rrrs rrKzHDFSClient._try_catsDWmmCQm77 V !88s## # rc t||z}t||z}|g|z}t|D]}||xxdz cc<gg|z}d} t|D]#}|| | ||z||<| ||z } $||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 r0rk trainer_idtrainers remainder blocksizeblocksr# trainer_filesbegins rrzHDFSClient._split_filesJJ) JJ(* x'y!!  A 1IIINIIIIx x  A$UUVAY->%>?M!  VAY EEZ((r path_listlist[_FileInfo]ct|dkrgSg}d}|D] }||dzz } d|zdz}||\}}t|dkrtjd|dgS|D]`}|d} t| dkr+| d } t | d} || | d a|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 warningrrri) r0rY file_list str_concatrrrrrr file_path file_sizes rlist_files_infozHDFSClient.list_files_infos y>>Q  I   % %D $* $JJ J !P P ]]3'' U u::?? NAYAAA B B BI E ED**S//C3xx!||AICF I   iCC D D D Drrr) rrrrrrrrrbrm)FrrrrF)rQrrRrrMr|rbrm)rF) r;rr1rr rrMr|rbrm) r1rr;rr rrMr|rbrmrFT) rHrrIrrMr|rNr|rbrmr]r)rbrEr,)r1rIrbr)rYrrbrZ)#rrrrrrrrrrUr2rrr7rr5r9rSr<r r>r*r@rOr:rurxrBrYrCrEr[rKrrcrrrrrs6 &      2(((( ))))<^B^%%%%B,^%%%%@   ))))@^!!!!H@E.....4 ! SSSSSj^    ! @@@@@D^   ^+(+(+(+(b  +6+6+6+6+6Z^   $$$ $$$ ^!!!!B%%%%%B^$$$      D^)))2      rrceZdZdZddZdZdZdZdZd Z d Z d Z d Z ddZ ddZddZdZddZdZddZdZd dZdZdS)! 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") rrcDtj|_||_dSr,)r AfsWrapper_fsr)r0rrs rrzAFSClient.__init__s?$$!rc@|j||||dSr,)rkinit)r0fs_namefs_user fs_passwdfs_confs rrmzAFSClient.inits"  gw 7;;;;;rcd||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: +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") rrs rrUzAFSClient.list_dirss7,}}W%% Ill7++ e rc^||sggfS||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: +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/s rr2zAFSClient.ls_dir8s3.}}W%% r6M||G$$$rcD|j|}|g}||fSr,)rklist)r0r1rkrjs rrzAFSClient._ls_dirTs& g&&yU{rcZ||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: +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/s rr7zAFSClient.is_dirYs/,}}W%% 5||G$$$rcd|j|}t|dkrdSdS)NrTF)rkrur)r0r1 list_paths rrzAFSClient._is_dirts0HMM'**  NNa  45rc\||sdS|| S)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/s rr5zAFSClient.is_file{s2,}}W%% 5<<((((rc6|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") )rkexistr/s rr9zAFSClient.is_exists.x~~g&&&rFc|d}|d}tj|}||dz|zr|r||dz|z||s|||j||dSr) r rfrrr9rBr@rkr<r s rrSzAFSClient.upload_dirs$$S)) ??3'')))44 ==C.8 9 9 9i 9 KK37 8 8 8}}X&& " KK ! ! !  8,,,,,rr ct}||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)rar9r$rkr<)r0r;r1r rMr s rr<zAFSClient.uploadsU. ~~j)) C&*'A'A'ABB B  G,,,,,rc||st|d||r|j||S||\}}|D]`}t j|t j |d}|j||adS)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", "./") r,r N) r9r$r5rkr>r2rfrrr) r0r1r;r rMr'r2 file_namelocal_file_names rr>zAFSClient.downloads.}}W%% ?&''='='=>> > <<  :8$$Z99 9 ;;w//=& : :I gllBGMM)44Q7O H  oy 9 9 9 9  : :rch||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)r9rkmkdirr/s rr@zAFSClient.mkdirss7& == ! !  F wrTc4|r*||r|||rN||st|d||rt|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") r8r9N)r9rBr$r"rkrOrLs rrOz AFSClient.mvs,  %{33 % KK $ $ $  I==-- K*k+I+I+IJJJ}}[)) I';(G(G(GHHH  K-----rch||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)r9rkrwr/s rrBzAFSClient.delete>s7(}}W%%  F      rcv||r |rdSt|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)r9r"rktouchzrWs rrYzAFSClient.touchVs?* == ! ! $ # #xw'''rcdSrGrrDs rrEzAFSClient.need_upload_downloadrrHrNcd||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)r5rkr[r/s rr[z AFSClient.catus1, <<  8<<(( (2rc t||z}t||z}|g|z}t|D]}||xxdz cc<gg|z}d} t|D]#}|| | ||z||<| ||z } $||SrNrOrPs rrzAFSClient._split_filesrXrrdre)r Frfr]r,)rrrrrrmrUr2rr7rr5r9rSr<r>r@rOrBrYrEr[rrrrrhrhs`$""""<<<8%%%8 %%%6)))6'''2----(----:#:#:#:#:J   . . . . .D!!!0((((86)))))rrhr,)rrrbr)' __future__rr^rrrfrrsrrtypingrrrrr paddle.baser log_utilr typing_extensionsr r rrrr__all__rrr"r$r&r(r*rarrrhrrrrs"#"""""   GGGGGGGGGGGGGG ++++++i ""G GG  E! LLI      9                9                   ;";";";";";";";"|jjjjjbjjj\ "&$$$$$Ne e e e e e e e Pe)e)e)e)e)e)e)e)e)e)r