【转】Python 访问 HDFS

1.前言

hdfs , Hadoop Distributed File System。Hadoop的分布式文件系统,安全行和扩展性没得说。

访问HDFS的方式有以下几种：

1. 命令行方式：FS Shell
2. 编程方式：FileSystem Java API，libhdfs(c语言)
3. REST API : WebHDFS, HttpFs
4. 把HDFS mount成本地文件目录

 

使用python访问HDFS比较容易：

• python 封装FS Shell， 这种方式本地必须部署了Hadoop 的 FS Shell。
• 使用python调用libhdfs，如果善于python和C语言可以采用此种方式，libhdfs是标准API的子集，有一些功能不能实现，网上有人说libhdfs比较坑
• python 封装REST API，支持windows环境
• 已有模块phdfs (封装了WebHDFS)，支持windows环境，类似的python模块还有HdfsCLI、webpyhdfs,pyhdfs
• snakebite，纯python hdfs client，使用了protobuf 和 hadoop rpc。

 

这里主要介绍使用hdfs 访问HDFS，支持python 2.7和 python 3

文档地址： hdfs 2.1.0

 

2.环境建立

1. Hadoop 使用已有的CDH 5.6.1 环境
2. 安装 hdfs，最新版本是0.2.1。

pip install hdfs

3. ipython, 可选但强烈建议。

3.示例代码

1. 创建目录并写文件

 

#!c:python27python.exe
# -*- coding: UTF-8 -*-
import traceback
from hdfs import InsecureClient
import time
import sys
from numpy import true_divide
reload(sys)
sys.setdefaultencoding("utf-8")


try:
    root_path = "/"
    #获取客户端，并且指定登陆用户，如果使用Client创建客户端就不能指定用户了
    c = InsecureClient(url="http://172.16.21.22:50070",user='hdfs',root=root_path)
    #创建目录
    c.makedirs('/user/root/pyhdfs')
    #写文件 
    #第一个参数：文件路径
    #第二个参数：文件内容
    #第三个参数：是否覆盖已有的文件，如果不覆盖，且文件已经存在会抛出异常
    c.write('/user/root/pyhdfs/1.log',time.asctime(time.localtime(time.time())) + '
',True)
    
    #下载文件
    #第一个参数，hdfs路径
    #第二个参数,本地路径
    #第三个参数,是否覆盖本地文件
    c.download('/user/root/pyhdfs/1.log', '.', True)
    
    #上传文件
    #第一个参数，hdfs路径
    #第二个参数,本地路径
    #第三个参数,是否覆盖hdfs已有的文件
    c.upload('/user/root/pyhdfs/', './pyhdfs_example.py', True)
    
    
    #获取目录下的的文件列表,第一个参数：hdfs路径，第二个参数是否获取文件的状态数据
    #另外pyhdfs 有一个walk函数 c.walk(hdfs_path, depth, status)
    #用法和os.walk类似，遍历目录非常好用
    hdfs_files = c.list('/user/root/pyhdfs', True)
    for f in hdfs_files:
        print f
    #输出结果如下
    #(u'1.log', {u'group': u'supergroup', u'permission': u'755', u'blockSize': 134217728, u'accessTime': 1519459768533L, u'pathSuffix': u'1.log', u'modificationTime': 1519459768569L, u'replication': 1, u'length': 25, u'childrenNum': 0, u'owner': u'hdfs', u'storagePolicy': 0, u'type': u'FILE', u'fileId': 33037})
    #(u'pyhdfs_example.py', {u'group': u'supergroup', u'permission': u'755', u'blockSize': 134217728, u'accessTime': 1519459768683L, u'pathSuffix': u'pyhdfs_example.py', u'modificationTime': 1519459768711L, u'replication': 1, u'length': 1624, u'childrenNum': 0, u'owner': u'hdfs', u'storagePolicy': 0, u'type': u'FILE', u'fileId': 33038})

    #返回文件的summary信息,算法是MD5-of-0MD5-of-512CRC32C
    #可以使用
    #返回结果：{u'spaceConsumed': 2567, u'quota': -1, u'spaceQuota': -1, u'length': 2567, u'directoryCount': 0, u'fileCount': 1}
    print c.content('/user/root/pyhdfs/pyhdfs_example.py')
    
    #返回的是md5的checksum
    #结果:{u'length': 28, u'bytes': u'0000020000000000000000005506d1dae3da4073662038f2392c08b700000000', u'algorithm': u'MD5-of-0MD5-of-512CRC32C'}
    #本地文件的checksum生成:hadoop fs -checksum /abc.txt
    print c.checksum('/user/root/pyhdfs/pyhdfs_example.py')
    #删除文件，第一个参数hdfs路径，第二参数，是否递归删除
    c.delete('/user/root/pyhdfs/', True)
    
    
    
except Exception , ex:
    print  traceback.format_exc()

 

4.其他hdfs Client api

• resolve(hdfs_path) ， 返回hdfs绝对路径
• status(hdfs_path,strict=True) 返回文件或者目录的状态
• acl_status（hdfs_path, strict=True), 返回acl的状态
• set_acl(hdfs_path, acl_spec, stict=True), 设置acl
• parts(hdfs_path, parts=None, status=False) ?
• read(hdfs_path, offset=0, length=None, buffer_size=None)
• rename(hdfs_src_path, hdfs_dest_path), 重命名
• set_owner(hdfs_paht, owner=None, group=None)
• set_permission(hdfs_path, permission)
• set_times(hdfs_path, access_time=None, modification_time=None)
• set_replication(hdfs_path, replication), 设置文件的副本数量, 只能设置文件，不能设置目录，否则会由异常
• walk(hdfs_path, depth=0, status=False), 类似 os.walk

上面都是基于 HDFS WEBAPI 进行的操作,记录一下。

Reference:

https://zhuanlan.zhihu.com/p/33983161   python访问HDFS