Hadoop（4）-Python借助pyhdfs对Hadoop的HDFS分布式文件系统的全部操作详细实例演示-摩杜云开发者社区

1、安装

执行如下命令即可安装

pip install pyhdfs

注意，在windows上对虚拟机上的HDFS进行远程操作时，需要在本机的hosts文件中填写ip和主机名的映射关系如这里：
Hadoop（4）-Python借助pyhdfs对Hadoop的HDFS分布式文件系统的全部操作详细实例演示

2、pyhdfs与HDFS常用的交互操作

在HDFS上指定目录下创建一个文件夹，然后查看此文件夹是否存在

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    fs.mkdirs("/test_01")
    file_or_dirs=fs.listdir("/")
    print(file_or_dirs)

执行结果为：

['test_01']

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    home_dir=fs.get_home_directory()
    print(home_dir)

执行结果如下：

/user/root

获取可用的节点

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    active_nodes=fs.get_active_namenode()
    print(active_nodes)

执行结果如下：这里因为是单机版环境

192.168.1.204:50070

在HDFS上创建一个文件，并写入内容：

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    all_files = fs.listdir("/")
    print(all_files)
    fs.create("/demo_01.txt",b"hello world",override=True)
    all_files=fs.listdir("/")
    print(all_files)

执行结果如下：

['test_01']
['demo_01.txt', 'test_01']

删除文件

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    all_files = fs.listdir("/")
    print(all_files)
    fs.delete("/demo_01.txt")
    all_files=fs.listdir("/")
    print(all_files)

执行结果如下：

['demo_01.txt', 'test_01']
['test_01']

查看文件是否存在

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    print(fs.exists("/demo_01.txt"))
    all_files = fs.listdir("/")
    print(all_files)
    fs.create("/demo_01.txt",b"hello world")
    print(fs.exists("/demo_01.txt"))
    all_files=fs.listdir("/")
    print(all_files)

执行结果如下：

False
['test_01']
True
['demo_01.txt', 'test_01']

读取文件内容：

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    f=fs.open("/demo_01.txt")
    ctx=f.read().decode("utf-8")
    print(ctx)
    f.close()

执行内容如下：

hello world

向文件中增加内容

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    f=fs.open("/demo_01.txt")
    ctx=f.read().decode("utf-8")
    print(ctx)
    f.close()
    fs.append("/demo_01.txt",b"\nhello hadoop!")
    print("-------------------------")
    f = fs.open("/demo_01.txt")
    ctx = f.read().decode("utf-8")
    print(ctx)
    f.close()

执行结果如下：

hello world
hello hadoop!
-------------------------
hello world
hello hadoop!
hello hadoop!

查看文件属性

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    status=fs.get_file_status("/demo_01.txt")
    print(status)

执行结果如下：

FileStatus(accessTime=1623756299241, blockSize=134217728, childrenNum=0, fileId=16389, group='supergroup', length=39, modificationTime=1623756729466, owner='root', pathSuffix='', permission='644', replication=1, storagePolicy=0, type='FILE')

查看当前文件的状态

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    status=fs.list_status("/demo_01.txt")
    print(status)

执行结果如下：

[FileStatus(accessTime=1623756299241, blockSize=134217728, childrenNum=0, fileId=16389, group='supergroup', length=39, modificationTime=1623756729466, owner='root', pathSuffix='', permission='644', replication=1, storagePolicy=0, type='FILE')]

list_status 和get_file_status的区别是list_status可以查看路径

重命名文件或文件夹

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    print(fs.listdir("/"))
    fs.rename("/test_01","/test_02")
    print(fs.listdir("/"))

执行结果如下：

['demo_01.txt', 'test_01']
['demo_01.txt', 'test_02']

将本地文件拷贝至HDFS

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    print(fs.listdir("/"))
    fs.copy_from_local("demo_02.txt","/demo_02.txt")
    print(fs.listdir("/"))

执行结果如下：

['demo_01.txt', 'test_02']
['demo_01.txt', 'demo_02.txt', 'test_02']

将HDFS上文件拷贝到本地

import pyhdfs

import os

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    print(os.listdir("."))
    fs.copy_to_local("/demo_01.txt","demo_01.txt")
    print(os.listdir("."))

执行结果如下：

['app1', 'db.sqlite3', 'demo_02.txt', 'manage.py', 'mysite1', 'static', 'templates', 'test.py']
['app1', 'db.sqlite3', 'demo_01.txt', 'demo_02.txt', 'manage.py', 'mysite1', 'static', 'templates', 'test.py']

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    summary=fs.get_content_summary("/")
    print(summary)

执行结果如下：

ContentSummary(directoryCount=2, ecPolicy='', fileCount=2, length=56, quota=9223372036854775807, snapshotDirectoryCount=0, snapshotFileCount=0, snapshotLength=0, snapshotSpaceConsumed=0, spaceConsumed=56, spaceQuota=-1, typeQuota={
     
       })

查看文件的校验和

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    check_sum=fs.get_file_checksum("/demo_01.txt")
    print(check_sum)

执行结果如下：

FileChecksum(algorithm='MD5-of-0MD5-of-512CRC32C', bytes='00000200000000000000000036e110bcf9265d7f5fdd153161b162d800000000', length=28)

设置文件所有者

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    print(fs.get_file_status("/demo_01.txt"))
    fs.set_owner("/demo_01.txt",owner="hdfs")
    print(fs.get_file_status("/demo_01.txt"))

执行结果如下：

FileStatus(accessTime=1623761641609, blockSize=134217728, childrenNum=0, fileId=16389, group='supergroup', length=39, modificationTime=1623756729466, owner='root', pathSuffix='', permission='644', replication=1, storagePolicy=0, type='FILE')
FileStatus(accessTime=1623761641609, blockSize=134217728, childrenNum=0, fileId=16389, group='supergroup', length=39, modificationTime=1623756729466, owner='hdfs', pathSuffix='', permission='644', replication=1, storagePolicy=0, type='FILE')

设置文件的副本数

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    print(fs.get_file_status("/demo_01.txt"))
    fs.set_replication("/demo_01.txt",replication=5)
    print(fs.get_file_status("/demo_01.txt"))

执行结果如下：

FileStatus(accessTime=1623761641609, blockSize=134217728, childrenNum=0, fileId=16389, group='supergroup', length=39, modificationTime=1623756729466, owner='root', pathSuffix='', permission='644', replication=1, storagePolicy=0, type='FILE')
FileStatus(accessTime=1623761641609, blockSize=134217728, childrenNum=0, fileId=16389, group='supergroup', length=39, modificationTime=1623756729466, owner='root', pathSuffix='', permission='644', replication=5, storagePolicy=0, type='FILE')

设置文件的修改时间和访问时间，修改时间和访问时间均为长整形，指从1970年1月1日开始的milliseconds

import pyhdfs

if __name__=="__main__":
    fs=pyhdfs.HdfsClient(hosts="192.168.1.204:50070",user_name="root")
    print(fs.get_file_status("/demo_01.txt"))
    fs.set_times("/demo_01.txt",modificationTime=1623756729499,accessTime=1623761641699)
    print(fs.get_file_status("/demo_01.txt"))

执行结果如下：

FileStatus(accessTime=1623761641609, blockSize=134217728, childrenNum=0, fileId=16389, group='supergroup', length=39, modificationTime=1623756729466, owner='root', pathSuffix='', permission='644', replication=5, storagePolicy=0, type='FILE')
FileStatus(accessTime=1623761641699, blockSize=134217728, childrenNum=0, fileId=16389, group='supergroup', length=39, modificationTime=1623756729499, owner='root', pathSuffix='', permission='644', replication=5, storagePolicy=0, type='FILE')

以上基本是pyhdfs的全部操作