Shulou(Shulou.com)06/03 Report--

[third Party package]

Pyhdfs (pypi,github, supports HA)

[function]

Rename a hdfs file or directory

# encoding: utf-8# author: walker# date: 2018-03-17 # summary: rename the hdfs file or directory import os, sys, timefrom pyhdfs import HdfsClientSrcPath ='/ test/xxx'DstPath ='/ test/yyy'NameNode = 'nn1.example.com:50070,nn2.example.com:50070'# using pyhdfs to rename SrcPath to DstPathdef Rename (SrcPath DstPath): fs = HdfsClient (hosts=NameNode) if not fs.exists (SrcPath): print ('Error: not found% s'% SrcPath) sys.exit (- 1) print ('Reanme.\ n% s\ n->\ n% s\ n'% (SrcPath, DstPath)) fs.rename (SrcPath DstPath) if _ _ name__ = ='_ main__': Rename (SrcPath, DstPath)

Upload files

# encoding: utf-8# author: walker# date: 2018-01-2 upload summary: upload local files to the hdfs directory import os, sys, timefrom pyhdfs import HdfsClientfrom configparser import ConfigParsercur_dir_fullpath = os.path.dirname (os.path.abspath (_ _ file__)) StartTime = time.time () FileSize = 0 # Total file size LocalDir =''HdfsDir =' 'NameNode =' UserName ='# # read configuration file def ReadConfig (): global LocalDir, HdfsDir, NameNode UserName cfg = ConfigParser () cfgFile = os.path.join (cur_dir_fullpath, 'config.ini') if not os.path.exists (cfgFile): input (cfgFile +' not found') sys.exit (- 1) cfgLst = cfg.read (cfgFile) if len (cfgLst)

< 1: input('Read config.ini failed...') sys.exit(-1) LocalDir = cfg.get('config', 'LocalDir').strip() if not os.path.exists(LocalDir): input(LocalDir + ' not found') sys.exit(-1) print('LocalDir:' + LocalDir) HdfsDir = cfg.get('config', 'HdfsDir').strip() print('HdfsDir:' + HdfsDir) NameNode = cfg.get('config', 'NameNode').strip() print('NameNode:' + NameNode) UserName = cfg.get('config', 'UserName').strip() print('UserName:' + UserName) print('Read config.ini successed!') #处理一个def ProcOne(client, srcFile, dstFile): global FileSize print('ProcOne \n%s\n ->

\ n% s'% (srcFile, dstFile)) # the target file already exists and has the same size if client.exists (dstFile) and\ (os.path.getsize (srcFile) = = client.list_status (dstFile) [0] .length): print ('file exists:% s'% dstFile) return True # Note If it already exists, it will be overwritten by client.copy_from_local (srcFile, dstFile Overwrite=True) # check file size if os.path.getsize (srcFile) = = client.list_status (dstFile) [0] .length: FileSize + = os.path.getsize (srcFile) return True return False # handles all def ProcAll (): client = HdfsClient (hosts=NameNode User_name=UserName) if not client.exists (HdfsDir): print (HdfsDir + 'not found') sys.exit (- 1) total = len (os.listdir (LocalDir)) processed = 0 failedList = list () for filename in os.listdir (LocalDir): srcFile = os.path.join (LocalDir) Filename) dstFile = HdfsDir +'/'+ filename if not ProcOne (client, srcFile, dstFile): failedList.append (srcFile) processed + = 1 print ('% d/%d/%d, time cost:% .2f s'% (total, processed, len (failedList)) Time.time ()-StartTime) print ('% DB,% .2f MB/s\ n'% (FileSize, FileSize/1024/1024/ (time.time ()-StartTime)) if failedList: print ('failedList:% s'% repr (failedList)) else: print ('Good! No errorations') Print ('% DB,% .2f MB,% .2f GB,% .2f MB/s'%\ (FileSize, FileSize/1024/1024, FileSize/1024/1024/1024) FileSize/1024/1024/ (time.time ()-StartTime)) if _ _ name__ = ='_ _ main__': ReadConfig () ProcAll () print ('Time total:% .2f s'% (time.time ()-StartTime)) print (time.strftime ('% Y-%m-%d% HRV% Mango% Scandinavian time.localtime ()

Download the HDFS file locally

# encoding: utf-8# author: walker# date: 2018-06-0cm summary: download the HDFS file (or directory) to the local import os, sys, timefrom pyhdfs import HdfsClientfrom configparser import ConfigParsercur_dir_fullpath = os.path.dirname (os.path.abspath (_ _ file__)) StartTime = time.time () FileSize = 0 # Total file size LocalDir =''HdfsDir =' 'NameNode =' UserName ='# read configuration file def ReadConfig (): global LocalDir, HdfsDir, NameNode UserName cfg = ConfigParser () cfgFile = os.path.join (cur_dir_fullpath, 'config.ini') if not os.path.exists (cfgFile): input (cfgFile +' not found') sys.exit (- 1) cfgLst = cfg.read (cfgFile) if len (cfgLst)

< 1: input('Read config.ini failed...') sys.exit(-1) LocalDir = cfg.get('config', 'LocalDir').strip() if not os.path.exists(LocalDir): input(LocalDir + ' not found') sys.exit(-1) print('LocalDir:' + LocalDir) HdfsDir = cfg.get('config', 'HdfsDir').strip().rstrip('/') print('HdfsDir:' + HdfsDir) NameNode = cfg.get('config', 'NameNode').strip() print('NameNode:' + NameNode) UserName = cfg.get('config', 'UserName').strip() print('UserName:' + UserName) print('Read config.ini successed!') #处理一个def ProcOne(client, srcFile, dstFile): global FileSize print('ProcOne \n%s\n ->

\ n% s'% (srcFile DstFile)) dstDir = os.path.dirname (dstFile) if not os.path.exists (dstDir): os.makedirs (dstDir) # the target file already exists and has the same size if os.path.exists (dstFile) and\ (os.path.getsize (dstFile) = = client.list_status (srcFile) [0] .length): print ('file exists:% s'% dstFile) return True # Note If it already exists, client.copy_to_local (srcFile, dstFile, overwrite=True) if os.path.getsize (dstFile)! = client.list_status (srcFile) [0] .length: # check file size return False FileSize + = os.path.getsize (dstFile) return True # handle all def ProcAll (): client = HdfsClient (hosts=NameNode User_name=UserName) if not client.exists (HdfsDir): print (HdfsDir + 'not found') sys.exit (- 1) total = 0 # go through it first Get the total number of files for parent, dirnames, filenames in client.walk (HdfsDir): for filename in filenames: total + = 1 processed = 0 failedList = list () for parent, dirnames, filenames in client.walk (HdfsDir): for filename in filenames: srcFile ='% s hand% s'% (parent, filename) relPath = srcFile [len (HdfsDir) + 1:] .replace ('/' The path relative to the root directory dstFile = os.path.join (LocalDir, relPath) if not ProcOne (client, srcFile, dstFile): failedList.append (srcFile) processed + = 1 print ('% d/%d/%d, time cost:% .2f s'% (total, processed, len (failedList)) Time.time ()-StartTime) print ('% DB,% .2f MB/s\ n'% (FileSize, FileSize/1024/1024/ (time.time ()-StartTime)) if failedList: print ('failedList:% s'% repr (failedList)) else: print ('Good! No errorations') Print ('% DB,% .2f MB,% .2f GB,% .2f MB/s'%\ (FileSize, FileSize/1024/1024, FileSize/1024/1024/1024) FileSize/1024/1024/ (time.time ()-StartTime)) if _ _ name__ = ='_ _ main__': ReadConfig () ProcAll () print ('Time total:% .2f s'% (time.time ()-StartTime)) print (time.strftime ('% Y-%m-%d% HRV% Mango% Scandinavian time.localtime ()

* walker * *

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.