ΑiSSKJr SSKrSSKrSSKrSSKrSSKrSSKrSSKrSSK r SSK J r J r J r JrJr SSKJr SSKJr \ (a-SSKJr \"S5r\"S 5r\"S \\S .5r"S S \5r/r"SS\5r"SS\5r"SS\5r"SS\5r "SS\5r!"SS5r""SS\"5r#S!S"Sjjr$"SS\"5r%"SS \"5r&g)#) annotationsN) TYPE_CHECKINGCallableLiteral TypedDictTypeVar)core)logger) ParamSpec_InputT_RetT_HDFSClientConfig)zfs.default.namezhadoop.job.ugic*\rSrSr%S\S'S\S'Srg) _FileInfo,strpathintsizeN)__name__ __module__ __qualname____firstlineno____annotations____static_attributes__ra/var/www/html/banglarbhumi/venv/lib/python3.13/site-packages/paddle/distributed/fleet/utils/fs.pyrr,s   rrc\rSrSrSrg) ExecuteError4rNrrrrrrrrr!r!4rr!c\rSrSrSrg)FSFileExistsError8rNr#rrrr&r&8r$rr&c\rSrSrSrg)FSFileNotExistsError<rNr#rrrr)r)<r$rr)c\rSrSrSrg) FSTimeOut@rNr#rrrr,r,@r$rr,c\rSrSrSrg)FSShellCmdAbortedDrNr#rrrr/r/Dr$rr/c<\rSrSr\R S5r\R S5r\R S5r\R S5r \R S5r \R S5r \R S5r \R S 5r \R S 5r\R S 5r\R SS j5r\R S 5r\R S5r\R SSj5r\R SSj5rSrg)FSHc[eNNotImplementedErrorselffs_paths rls_dir FS.ls_dirI!!rc[er5r6r8s ris_file FS.is_fileMr=rc[er5r6r8s ris_dir FS.is_dirQr=rc[er5r6r8s ris_exist FS.is_existUr=rc[er5r6)r9 local_pathr:s rupload FS.uploadYr=rc[er5r6)r9r:rHs rdownload FS.download]r=rc[er5r6r8s rmkdirs FS.mkdirsar=rc[er5r6r8s rdelete FS.deleteer=rc[er5r6r9s rneed_upload_downloadFS.need_upload_downloadir=rc[er5r6r9 fs_src_path fs_dst_paths rrename FS.renamemr=rc[er5r6r9rZr[ overwrite test_existss rmvFS.mvqr=rc[er5r6)r9 local_dirdest_dirs r upload_dir FS.upload_dirur=rc[er5r6r8s r list_dirs FS.list_dirsyr=rc[er5r6r9r:exist_oks rtouchFS.touch}r=rNc[er5r6r8s rcatFS.catr=rrFFTr5)rrrrabcabstractmethodr;r?rBrErIrLrOrRrVr\rbrgrjrorrrrrrr2r2Hs"" "" "" "" "" "" "" "" "" "" "" "" "" "" ""rr2c\rSrSrSrSSjrSSjrSSjrSrSr SSjr SS jr SS jr SS jr SS jrSSS jjrSSSjjrSSjrSrg)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("./") cURU5(d//4$/n/n[R"U5HQn[RR US-U-5(aUR U5 M@UR U5 MS X#4$)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("./") /)rEoslistdirrisdirappend)r9r:dirsfilesfs rr;LocalFS.ls_dirsu*}}W%%r6MG$Aww}}Ws]Q.// A Q % {rc[RRU5(a US35e[R"USS9 g)aY 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)rnN)r}risfilemakedirsr8s rrOLocalFS.mkdirss8$77>>'**Jwi7I,JJ* Gd+rc0[R"X5 g)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 >>> # 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\rYs rr\LocalFS.renames4 ++rc0[R"U5 gr5)shutilrmtreer8s r_rmr LocalFS._rmrs grc0[R"U5 gr5)r}remover8s r_rm LocalFS._rms 'rcURU5(dg[RRU5(aUR U5$UR U5$)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)rEr}rrrrr8s rrRLocalFS.deletesG$}}W%%  77>>' " "88G$ $yy!!rcg)NFrrUs rrVLocalFS.need_upload_downloadsrc@[RRU5$)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 >>> # 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}rrr8s rr?LocalFS.is_file s.ww~~g&&rc@[RRU5$)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}rrr8s rrBLocalFS.is_dir$s.ww}}W%%rc@[RRU5$)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") )r}rexistsr8s rrELocalFS.is_exist=s*ww~~g&&rcURU5(aU(ag[e[US5 SSS5 g!,(df  g=f)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 >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.distributed.fleet.utils import LocalFS >>> client = LocalFS() >>> client.touch("test_touch") >>> client.delete("test_touch") Na)rEr&openrms rro LocalFS.touchTs8( == ! !# # '3    s < A cURU5(d[eU(a'URU5(aURU5 URU5(a[eUR X5$)ap 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") )rEr)rRr&r\)r9src_pathdst_pathr`ras rrb LocalFS.mvosZ6}}X&&& & x00 KK ! == " "# #{{8..rcURU5(d/$[R"U5Vs/sH1n[RR US-U-5(dM/UPM3 nnU$s snf)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("./") r|)rEr}r~rr)r9r:rrs rrjLocalFS.list_dirssd(}}W%%Izz'* *!bggmmGcMA ! $/$/$/ $/  $/  $/Lrryc^SU4SjjnU$)NcN>^[R"T5SUU4Sjj5nU$)NcZ>USnT nUc[UR5S- nOUS-n[UR5S- n[R"5nUnT"U0UD6$![acn[R"5U- U:a'[ SUS[R"5U- 35e[R "U5 SnAOSnAff=f[R"5U- S:a<[SUS[R"5U- 35 [R"5nM)Nrg@@zargs:z timeout:zhadoop operator timeout:args:)float _time_out _sleep_intertimer!r,sleepprint) argskwargsotime_outinterstartlast_print_timeer max_time_outs rhandler2_handle_errors..decorator..handlersQA#H -6F"!..)F2EIIKE#O &d-f--#&yy{U*h6'#D6499;3F2GHJJu%% &99;0257vYtyy{UZGZF[\'+iikO!sA## C-AC  C)rz _InputT.argsrz_InputT.kwargsrr) functoolswraps)rrrs` r decorator!_handle_errors..decorators%   2  2:r)rCallable[_InputT, _RetT]rrr)rrs` r_handle_errorsrsB rcz\rSrSr%SrS\S'S$S%SjjrS&SjrS&Sjr\ "5S'Sj5r \ "5S(S j5r S r S r \ "5S)S j5rS rS)Sjr\ "5S)Sj5rS*S+SjjrS,S-Sjjr\ "5S5rS,S.Sjjr\ "5S5r\ "5S/Sj5rS0S1Sjjr\ "5S5rSrSr\ "5S/Sj5rS2S3Sjjr\ "5S5rS4SjrS5S6Sjjr \ "5S 5r!S!r"S7S"jr#S#r$g)8 HDFSClientia 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_commandsc/UlUS3nURRU5 SnURRU5 U(a<UR5H(upxSUSU3n URRU 5 M* X0lX@lSR UR5Ul[R"S5Ul g)Nz /bin/hadoopfsz-D= z8\s?responseErrorMsg\s?\:.*, errorCode\:\s?[0-9]+, path\:) rritemsrrjoin _base_cmdrecompile _bd_err_re) r9 hadoop_homeconfigsr sleep_inter hadoop_bindfskvconfig_commands r__init__HDFSClient.__init__s#}K0    ,   %  #%aS!!!((8("'$"3"34** G rc&URSU3nSnSnSn[US-5HGn[R"USSU5upV[ U5nUS:Xa O[ R "U5 MI US:Xa [U5eXVR54$)Nz -rr rz) rranger shell_execute_cmdrrrr/ splitlines) r9cmdredirect_stderr retry_timesexe_cmdretoutputretry_sleep_secondxs r_run_cmdHDFSClient._run_cmds^^$Bse,{Q'A00!QPKCc(Cax JJ) * ( #:#C( (%%'''rc UR/UR5QnSnSnSn[US-5H[n[R"US[R U(a[R O[R SS9n U Rn O US:Xa [U5eg![Ra9n U RnU Rn[R"U5 Sn A MSn A f[a n Sn A MnSn A ff=f)Nrrr T)checkstdoutstderrtextrz)rsplitr subprocessrunPIPESTDOUTrCalledProcessError returncoderrr Exceptionr/) r9rrrrrrrrprocessrs r _run_safe_cmdHDFSClient._run_safe_cmd!s>>0CIIK0{Q'A $..%??+#))'__ !(, #:#C( ( 00 /ll -..  sAB##D7.C++ D?Dc\URU5(d/$URU5up#U$)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") rE_ls_dirr9r:rrs rrjHDFSClient.list_dirs?s-8}}W%%Ill7+  rcXURU5(d//4$URU5$)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") rr8s rr;HDFSClient.ls_diras+:}}W%%r6M||G$$rcZSU3nURU5up4US:wa [U5e/n/nUHvnUR5n[U5S:waM$[R R US5n USSS:XaURU 5 MeURU 5 Mx XV4$)Nls rd)rr!rlenr}rbasenamer) r9r:rrlinesrrlinearrps rrHDFSClient._ls_dirsG9o]]3'  !8s# #D**,C3x1}  Q(A1vayC A Q{rcZUH%nURRU5nUcM#Us $ gr5)rmatch)r9rlms r _test_matchHDFSClient._test_matchs/A%%a(A} rcRURU5(dgURU5$)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") FrE_is_dirr8s rrBHDFSClient.is_dirs%8}}W%%||G$$rcSU3nURUSSS9up4U(aGURU5(a0[S5 [SRU55 [ U5egg)Nztest -d Tr rrzraise exception:  F)rrrrr!)r9r:rrrs rr HDFSClient._is_dirsc "]]3!]L  &&)*dii&'"3''rc\URU5(dgURU5(+$)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") Frr8s rr?HDFSClient.is_files(6}}W%%<<(((rcDSUS3nURUSSS9up4US:wagg)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: +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 r#rF)r)r9r:rrouts rrEHDFSClient.is_exists5: #==d=J !8rcjURS5nURS5n[RRU5nUR US-U-5(aU(aUR US-U-5 UR U5(dUR U5 URX5 gz 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}rrrErRrO _try_uploadr9rerfr`local_basenames rrgHDFSClient.upload_dirs$$S) ??3')))4 ==C.8 9 9i KK37 8}}X&& KK ! -rc^U4SjnSn[5nURU5(d[US35eU"U5nU(d [S5 gTRU5(a)U(a"TR U5 TR U5 /n [ U5HLn TRXU5n [R"XRU 4S9n U RU 5 U R5 MN U Hn U R5 M g)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:>UHnTRX 5 M gr5)r.)hdfs_path_singledatasdatar9s r__subprocess_upload.HDFSClient.upload..__subprocess_uploadLs  8rc^/n[RRU5(dU$[RRU5(aO[R"U5H3n[RR X5nUR U5 M5 U$UR U5 U$)zR get local files Args: path(str): local path Returns: list of local files )r}rrrr~rr)rrlistfilets rget_local_files*HDFSClient.upload..get_local_filesPs~E77>>$'' ww}}T""JJt,D T0ALLO- L T"Lr not existsz/there are nothing need to upload, function exitNtargetr) ryrEr)rrRrOr _split_filesmultiprocessingProcessrrr)r9rHr:multi_processesr`_HDFSClient__subprocess_uploadr=local all_filesprocsi process_datasrprocs` rrIHDFSClient.upload*sD 9 * ~~j))&*['AB B#J/  C D  == ! !i KK KK 'A --iOLM''*=1IA LLO GGI (D IIKrcSUSU3nSnURU5upEUS:wa [U5eg![anURU5 UeSnAff=f)Nzput rr)rr!rrR)r9rHr:rr_rs rr.HDFSClient._try_uploadsfZL'+ ]]3'FCax"3''  KK G s$1 AAAcJ^U4SjnTRU5(d[US35eTRU5(aTRX5$TR U5upgUVs/sH oS-U-PM n nU R UVs/sH oS-U-PM sn5 /n [ U5HLnTRXU5n [R"XRU 4S9n U RU 5 U R5 MN U Hn U R5 M gs snfs snf)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 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:>UHnTRX 5 M g)zp download file from HDFS Args: local_path(str): the local file path datas(str): the hdfs file path list N) _try_download)rHr5r6r9s r__subprocess_download2HDFSClient.download..__subprocess_downloads""44r not exitsr|r@N) rEr)r?rSr;extendrrBrCrDrrr)r9r:rHrEr` _HDFSClient__subprocess_downloadr all_filenamesrJrHrIrKrrLs` rrLHDFSClient.downloads D 5}}W%%&'*'=> > << %%g: :"kk'20=> 1s]Q& >T:TC-!+T:;'A --iOLM'', 3NA LLO GGI (D IIK?:s *D D cSUSU3nSnURU5upEUS:wa [U5eg![a"n[5nUR U5 UeSnAff=f)Nzget rr)rr!rryrR)r9r:rHrrrOrlocal_fss rrSHDFSClient._try_downloadsmWIQzl+ ]]3'FCax"3'' yH OOJ 'G s$1 AAAcLURU5(agSnSUS3nURUSS9upEUS:wa%UH nSU;dM Sn O U(d [U5eU(aAURU5(d*S U3nURU5upGUS:wa [U5eggg) 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 )rErr!)r9r:out_hdfsrrr)rrOs rrOHDFSClient.mkdirss2 == ! ! wiq!==d=; !8.!3#H"3'' DMM'22gY'C]]3'FCax"3''38rcU(a'URU5(aURU5 U(aHURU5(d[US35eURU5(a[US35eUR X5$)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)rErRr)r&_try_mvr_s rrb HDFSClient.mv svB {33 KK $ ==--*k].+IJJ}}[))';-(GHH||K55rcSUSU3nSnURUSS9upEUS:wa [U5eg![a8nURU5(dURU5(aSnAgUeSnAff=f)Nzmv rrr r)rr!rrE)r9rZr[rrrOrs rrdHDFSClient._try_mv6s~K=+/ ]]3A]6FCax"3'' ==--$-- 2L2LG s#0 A2,A-+A--A2cVSU3nURU5up4US:wa [U5eg)Nzrmr rrr!r9r:rrrOs rrHDFSClient._rmrCs5WIs# !8s# # rcVSU3nURU5up4US:wa [U5eg)Nzrm rrjrks rrHDFSClient._rmIs4G9os# !8s# # rcURU5(dgURU5nU(aURU5$URU5$)aN 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)rEr rr)r9r:rBs rrRHDFSClient.deleteOsE2}}W%% g& 99W% %xx  rclURU5(aU(ag[eURU5$)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 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)rEr&_touchzrms rroHDFSClient.touchqs.4 == ! !# #||G$$rcVSU3nURU5up4US:wa [U5eg)Nztouchz rrjrks rrrHDFSClient._touchzs5y!s# !8s# # rcgNTrrUs rrVHDFSClient.need_upload_downloadrNctURU5(a"URU5nSRU5$g)aQ 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") '' r$r)r?_try_catr)r9r:rs rrrHDFSClient.cats38 << ]]7+F99V$ $rcVSU3nURUSS9up4US:wa [U5eU$)Nzcat r rgrrj)r9r:rrrs rr{HDFSClient._try_cats9WImmCQm7  !8s# # rc[U5U-n[U5U-nU/U-n[U5HnXg==S- ss'M //U-nSn [U5HnXXU-X'XU- n M X$z 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 r9r trainer_idtrainers remainder blocksizeblocksrJ trainer_filesbegins rrBHDFSClient._split_filesJ) J(* x'y!A INI"x xA$UAY->?M  AY E!((rc~[U5S::a/$/nSnUH nX4S-- nM SU-S-nURU5upg[U5S:Xa[R"SUS35 /$UHKnUR S5n [U 5S:aM%U S n [ U S5n UR XS .5 MM U$) z list_files return file path and size Args: path_list(list): file list Returns: filelist(list): file list with file path and size rrrr z) | awk '{if ($8 != "") {print $5" "$8 }}'zlist_files empty, path[]r )rr)rrr warningrrr) r9 path_list file_list str_concatrrrrrr file_path file_sizes rlist_files_infoHDFSClient.list_files_infos y>Q I  D * $J J !P P ]]3'  u:? NN4YKqA BID**S/C3x!|AICF I   iC D r)rrrrrii) rrrrrrrrrr)FrrrF)rerrfrr`rrr)rF) rHrr:rrErr`rrr) r:rrHrrErr`rrrrFT) rZrr[rr`rrarrrrur)rz Literal[True]r5)r:z str | Nonerr)rrrzlist[_FileInfo])%rrrrrrrrrrrjr;rrrBr r?rErgrIr.rLrSrOrbrdrrrRrorrrVrrr{rBrrrrrrrs6 &   #       2( )<B%%B,%%@ )@!!H@E..(+.8<. .4 ! SSS S  S  Sj   ! @@@ @  @  @D  +(+(b  +6+6+6 +6  +6  +6Z  $ $ !!B%B$$  D)2 rrc\rSrSrSrSSjrSrSrSrSr Sr S r S r S r SS jrSS jrSSjrSrSSjrSrSSjrSrSSjrSrSrg) AFSClientia 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") cD[R"5UlXlgr5)r AfsWrapper_fsr)r9rrs rrAFSClient.__init__s??$!rc<URRXX45 gr5)rinit)r9fs_namefs_user fs_passwdfs_confs rrAFSClient.inits  g ;rc\URU5(d/$URU5up#U$)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 rrjAFSClient.list_dirss-,}}W%%Ill7+  rcXURU5(d//4$URU5$)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") rr8s rr;AFSClient.ls_dir8s+.}}W%%r6M||G$$rcDURRU5nU/nX24$r5)rlist)r9r:rrs rrAFSClient._ls_dirTs# g&y{rcRURU5(dgURU5$)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") Frr8s rrBAFSClient.is_dirYs%,}}W%%||G$$rcZURRU5n[U5S:agg)NrTF)rrr)r9r: list_paths rr AFSClient._is_dirts&HHMM'*  Na rc\URU5(dgURU5(+$)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: +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") Frr8s rr?AFSClient.is_file{s(,}}W%%<<(((rc8URRU5$)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") )rexistr8s rrEAFSClient.is_exists.xx~~g&&rc~URS5nURS5n[RRU5nUR US-U-5(aU(aUR US-U-5 UR U5(dUR U5 URRX5 gr,) r-r}rrrErRrOrrIr/s rrgAFSClient.upload_dirs$$S) ??3')))4 ==C.8 9 9i KK37 8}}X&& KK !  ,rc[5nURU5(d[US35eURR X5 g)aj 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") r?N)ryrEr)rrI)r9rHr:rEr`rGs rrIAFSClient.uploads<. ~~j))&*['AB B  ,rcURU5(d[US35eURU5(aURR X!5$UR U5upVUH^n[ RRU[ RRU5S5nURR X5 M` g)aG 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", "./") rVr N) rEr)r?rrLr;r}rrr) r9r:rHrEr`rOrY file_namelocal_file_names rrLAFSClient.downloads.}}W%%&'*'=> > << 88$$Z9 9 ;;w/&I ggllBGGMM)4Q7O HH  o 9 'rchURU5(agURRU5 g)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)rErmkdirr8s rrOAFSClient.mkdirss&& == ! !  wrc4U(a'URU5(aURU5 U(aHURU5(d[US35eURU5(a[US35eURR X5 g)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") rbrcN)rErRr)r&rrbr_s rrb AFSClient.mvsv, {33 KK $ ==--*k].+IJJ}}[))';-(GHH  K-rchURU5(dgURRU5 g)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)rErrr8s rrRAFSClient.delete>s&(}}W%%   rcURU5(aU(ag[eURRU5$)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)rEr&rtouchzrms rroAFSClient.touchVs2* == ! !# #xxw''rcgrwrrUs rrVAFSClient.need_upload_downloadrryrNcfURU5(aURRU5$g)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?rrrr8s rrr AFSClient.catus), << 88<<( (rc[U5U-n[U5U-nU/U-n[U5HnXg==S- ss'M //U-nSn [U5HnXXU-X'XU- n M X$rrrs rrBAFSClient._split_filesrr)rrrr)r Frrur5)rrrrrrrrjr;rrBr r?rErgrIrLrOrbrRrorVrrrBrrrrrrsh$"<8%8 %6)6'2-(-:#:J . .D!0(86)rrr5)rz float | Nonerz>Callable[[Callable[_InputT, _RetT]], Callable[_InputT, _RetT]])' __future__rrvrrCr}rrrrtypingrrrrr paddle.baser log_utilr typing_extensionsr r rrrr__all__rr!r&r)r,r/r2ryrrrrrrrs"#   GG+ "G G E!LI  9    9     ;";"|jbj\ "&$$C$Ne e Pe)e)r