python使用hdfs3模塊對hdfs進行操作詳解

之前一直使用hdfs的命令進行hdfs操作，比如：

  hdfs dfs -ls /user/spark/  hdfs dfs -get /user/spark/a.txt /home/spark/a.txt #從HDFS獲取數據到本地  hdfs dfs -put -f /home/spark/a.txt /user/spark/a.txt #從本地覆蓋式上傳  hdfs dfs -mkdir -p /user/spark/home/datetime=20180817/  ....

身為一個python程序員，每天操作hdfs都是在程序中寫各種cmd調用的命令，一方面不好看，另一方面身為一個Pythoner這是一個恥辱，於是乎就挑了一個hdfs3的模塊進行hdfs的操作，瞬間就感覺優雅多了：
 

hdfs 官方API：https://hdfs3.readthedocs.io/en/latest/api.html

  >>> from hdfs3 import HDFileSystem  #鏈接HDFS  >>> hdfs = HDFileSystem(host='localhost', port=8020)  >>> hdfs.ls('/user/data')  >>> hdfs.put('local-file.txt', '/user/data/remote-file.txt')  >>> hdfs.cp('/user/data/file.txt', '/user2/data')

#文件讀取

  #txt文件全部讀取  >>> with hdfs.open('/user/data/file.txt') as f:  ...  data = f.read(1000000)  #使用pandas讀取1000行數據  >>> with hdfs.open('/user/data/file.csv.gz') as f:  ...  df = pandas.read_csv(f, compression='gzip', nrows=1000)

#寫入文件

>>> with hdfs.open('/tmp/myfile.txt', 'wb') as f:
 ... f.write(b'Hello, world!')

#多節點連接設置

  host = "nameservice1"  conf = {"dfs.nameservices": "nameservice1",    "dfs.ha.namenodes.nameservice1": "namenode113,namenode188",    "dfs.namenode.rpc-address.nameservice1.namenode113": "hostname_of_server1:8020",    "dfs.namenode.rpc-address.nameservice1.namenode188": "hostname_of_server2:8020",    "dfs.namenode.http-address.nameservice1.namenode188": "hostname_of_server1:50070",    "dfs.namenode.http-address.nameservice1.namenode188": "hostname_of_server2:50070",    "hadoop.security.authentication": "kerberos"  }  fs = HDFileSystem(host=host, pars=conf)

#API

  hdfs = HDFileSystem(host='127.0.0.1', port=8020)  hdfs.cancel_token(token=None) #未知，求大佬指點  hdfs.cat(path) #獲取指定目錄或文件的內容  hdfs.chmod(path, mode) #修改制定目錄的操作權限  hdfs.chown(path, owner, group) #修改目錄所有者，以及用戶組  hdfs.concat(destination, paths) #將指定多個路徑paths的文件，合併成一個文件寫入到destination的路徑，並刪除源文件（The source files are deleted on successful completion.成功完成後將刪除源文件。）  hdfs.connect() #連接到名稱節點 這在啟動時自動發生。 LZ:未知作用，按字面意思，應該是第一步HDFileSystem(host='127.0.0.1', port=8020)發生的  hdfs.delegate_token(user=None)  hdfs.df() #HDFS系統上使用/空閒的磁盤空間  hdfs.disconnect() #跟connect()相反，斷開連接  hdfs.du(path, total=False, deep=False) #查看指定目錄的文件大小，total是否把大小加起來一個總數，deep是否遞歸到子目錄  hdfs.exists(path) #路徑是否存在  hdfs.get(hdfs_path, local_path, blocksize=65536) #將HDFS文件複製到本地,blocksize設置一次讀取的大小  hdfs.get_block_locations(path, start=0, length=0) #獲取塊的物理位置  hdfs.getmerge(path, filename, blocksize=65536) #獲取制定目錄下的所有文件，複製合併到本地文件  hdfs.glob(path) #/user/spark/abc-*.txt 獲取與這個路徑相匹配的路徑列表  hdfs.head(path, size=1024) #獲取指定路徑下的文件頭部分的數據  hdfs.info(path) #獲取指定路徑文件的信息  hdfs.isdir(path) #判斷指定路徑是否是一個文件夾  hdfs.isfile(path) #判斷指定路徑是否是一個文件  hdfs.list_encryption_zones() #獲取所有加密區域的列表  hdfs.ls(path, detail=False) #返回指定路徑下的文件路徑，detail文件詳細信息  hdfs.makedirs(path, mode=457) #創建文件目錄類似 mkdir -p  hdfs.mkdir(path) #創建文件目錄  hdfs.mv(path1, path2) #將path1移動到path2  open(path, mode='rb', replication=0, buff=0, block_size=0) #讀取文件，類似於python的文件讀取  hdfs.put(filename, path, chunk=65536, replication=0, block_size=0) #將本地的文件上傳到，HDFS指定目錄  hdfs.read_block(fn, offset, length, delimiter=None) #指定路徑文件的offset指定讀取字節的起始點，length讀取長度，delimiter確保讀取在分隔符bytestring上開始和停止  >>> hdfs.read_block('/data/file.csv', 0, 13)   b'Alice, 100 Bo'  >>> hdfs.read_block('/data/file.csv', 0, 13, delimiter=b' ')   b'Alice, 100 Bob, 200'  hdfs.rm(path, recursive=True) #刪除指定路徑recursive是否遞歸刪除  hdfs.tail(path, size=1024) #獲取 文件最後一部分的數據  hdfs.touch(path) #創建一個空文件  hdfs.walk(path) #遍歷文件樹

補充知識：HDFS命令批量創建文件夾和文件

批量創建測試文件夾：

hadoop fs -mkdir -p /user/hivedata/temp/201912311001/d={27..30}/h={10..17}/m5={5,15,25}/

批量創建測試文件：

hadoop fs -touchz /user/hivedata/temp/201912311001/d={27..30}/h={10..17}/m5={5,15,25}/{0..5}.orc

最終效果：

hadoop fs -ls -R /user/hivedata/