import pyhdfs
from pyhdfs import HdfsException
import os
import telnetlib as tn
from flask import Flask,request,Response
from gevent.pywsgi import WSGIServer
import werkzeug
'''
python 操作HDFS。
1.访问文件系统列表（目录、文件）
FileStatus(accessTime=0,
blockSize=0,
childrenNum=15,
fileId=16395,
group='supergroup',
length=0,
modificationTime=1572431013356,
owner='hdfs',
pathSuffix='.cloudera_health_monitoring_canary_files',
permission='0',
replication=0,
storagePolicy=0,
type='DIRECTORY'/'FILE')
2.读取文件内容
3.下载HDFS文件到本地
'''
namenodes = 'ip:9870,ip:9870'
file_type = 'FILE'
dir_type = 'DIRECTORY'
app = Flask(__name__)
@app.route('/uploadBytes',methods=['POST'])
def upload_bytes():
method = request.method
if method != 'GET' and method != 'POST':
return Response({0: '请使用GET或POST请求！'})
# 获取request请求信息
# args = request.args #args为路径中的参数。
file = request.files['file']
dicts = request.form.to_dict()
dir = dicts['dirPath']
nodes = dicts['nameNodes']
filename = dicts['fileName']
# 上传到hdfs
myhdfs = MyHdfs(nodes)
result = myhdfs.upload_to_hdfs2(file, filename, dir)
return Response(str(result))
@app.route('/upload',methods=['POST'])
def upload():
method = request.method
if method != 'GET' and method != 'POST':
return Response({0:'请使用GET或POST请求！'})
#获取request请求信息
#args = request.args #args为路径中的参数。
file = request.files['file']
dicts = request.form.to_dict()
dir = dicts['dirPath']
nodes = dicts['nameNodes']
#上传到hdfs
myhdfs = MyHdfs(nodes)
result = myhdfs.upload_to_hdfs(file,dir)
return Response(str(result))
@app.route('/download',methods=['GET'])
def download():
if request.method != 'GET':
return Response({0:'请使用GET请求！'})
args = request.args
nodes = args.get("nameNodes")
file = args.get("filePath")
# 连接到hdfs
myhdfs = MyHdfs(nodes)
result = myhdfs.get_hdfs_filestream(file)
if type(result) == dict:
return Response(str(result))
return Response(result)
@app.route('/read',methods=['GET'])
def readfile():
if request.method != 'GET':
return Response({0:'请使用GET请求！'})
args = request.args
nodes = args.get("nameNodes")
file = args.get("filePath")
lines = args.get("readLines")
# 连接到hdfs
myhdfs = MyHdfs(nodes)
result = myhdfs.read_hdfs_file(file,lines)
return Response(str(result))
@app.route('/delete',methods=['GET'])
def delfile():
if request.method != 'GET':
return Response({0:'请使用GET请求！'})
args = request.args
nodes = args.get("nameNodes")
file = args.get("filePath")
# 连接到hdfs
myhdfs = MyHdfs(nodes)
result = myhdfs.delete_file(file)
return Response(str(result))
class PathInfo(object):
"""存储路径（目录、文件）信息"""
def __init__(self, name, pathInfos=[]):
#pathInfo自身的路径名称
self.name = name
#pathInfo下一级的路径名称
self.pathInfos = pathInfos
def __str__(self):
return "{'name':'%s','sub':%s}"%(self.name,self.pathInfos)
class MyHdfsException(HdfsException):
def __init__(self, value="请检查主机、端口的正确性！"):
self.value = value
def raise_exception(self):
try:
raise MyHdfsException
finally:
print(self.value)
class MyHdfs(object):
def __init__(self, hosts,user_name='hdfs'):
self.hosts = hosts
self.username = user_name
self.fs = pyhdfs.HdfsClient(self.hosts,self.username)
self.__check_server_available()
def __check_server_available(self):
if len(self.hosts.strip())==0:
MyHdfsException().raise_exception()
try:
host_list = str(self.hosts).split(",")
for host in host_list:
ip,port = host.split(":")
tn.Telnet(ip, port)
except:
MyHdfsException().raise_exception()
else:
print('IP和端口可用!')
def upload_to_hdfs(self,data,dirPath="/"):
"""
上传文件到HDFS,默认上传到HDFS根目录。
:param data 需要上传的数据。(``bytes`` or a ``file``-like object)
:param dirPath HDFS目录。(可以是已经存在的目录，也可以是不存在的)
:return: 上传成功1，上传失败0
"""
try:
# 获取文件名
filename = data.filename
print("filename: ", filename)
#拼接文件路径
filePath = ''
if dirPath.endswith("/"):
filePath = dirPath + filename
else:
filePath = dirPath + '/' + filename
#检查文件是否存在
if self.fs.exists(filePath):
print("文件已存在！")
return {0:"文件已存在！"}
else:
# 创建文件
print("开始上传文件...")
self.fs.create(filePath, data, overwrite=False, buffersize=2048)
print("上传文件完成...")
return {1:"成功上传文件！"}
except:
#MyHdfsException("上传文件失败！").raise_exception()
print("上传文件失败！")
return {0:"上传文件失败！"}
def upload_to_hdfs2(self,data,filename,dirPath="/"):
"""
上传文件到HDFS,默认上传到HDFS根目录。
:param data 需要上传的数据。(``bytes`` or a ``file``-like object)
:param fileName 文件名
:param dirPath HDFS目录。(可以是已经存在的目录，也可以是不存在的)
:return: 上传成功1，上传失败0
"""
try:
#获取文件名
print("filename: ", filename)
#拼接文件路径
filePath = ''
if dirPath.endswith("/"):
filePath = dirPath + filename
else:
filePath = dirPath + '/' + filename
#检查文件是否存在
if self.fs.exists(filePath):
print("文件已存在！")
return {0:"文件已存在！"}
else:
# 创建文件
print("开始上传文件...")
self.fs.create(filePath, data, overwrite=False,buffersize=2048)
print("上传文件完成...")
return {1:"成功上传文件！"}
except:
#MyHdfsException("上传文件失败！").raise_exception()
print("上传文件失败！")
return {0:"上传文件失败！"}
def get_hdfs_filestream(self,filePath):
"""
获取指定文件的IO流
:param filePath: HDFS文件的路径
:return: 1表示获取文件流成功；0表示有问题
"""
try:
check_result = self.__check_hdfs_path(filePath,file_type)
if not check_result:
return {0: "请指定文件路径！"}
# 返回文件流:urllib3.response.HTTPResponse
return self.fs.open(filePath)
except MyHdfsException:
return {0:"文件路径不存在！"}
def download_to_local(self,hdfsSource,localDest):
"""
将HDFS上的文件下载到本地指定目录
:param hdfsSource: HDFS上的文件路径
:param localDest: 本地文件/目录路径
###param over_write:是否覆盖已经存在的文件
:return: 提示信息，或下载成功的文件数量n或失败0
"""
# 判断是否是文件
if self.__check_hdfs_path(hdfsSource,file_type):
#检查本地目录：如果不存在,就创建
localDest,localDest_type = self.check_local_path(localDest)
# 直接下载
if localDest_type == dir_type:
idx = hdfsSource.rindex("/")
localDest = localDest + "/" + hdfsSource[idx:]
self.fs.copy_to_local(hdfsSource, localDest)
return 1
else:
print("HDFS路径必须是文件路径！")
return 0
def __check_hdfs_path(self,hdfsPath,isFileOrDir):
"""
判断给定的HDFS是否存在，如果存在，是文件还是目录。
:param hdfsPath: 路径
:param isFileOrDir: 文件或目录,取值为'DIRECTORY'或'FILE'
:return:
"""
if (not self.fs.exists(hdfsPath)):
MyHdfsException("路径(%s)不存在！" % (hdfsPath)).raise_exception()
file_status = self.fs.get_file_status(hdfsPath)
return file_status.type == isFileOrDir
def check_local_path(self,givenpath):
"""
判断文件或目录是否存在，不存在就创建目录
:param givenpath:
:return: 返回给定的givenpath是文件还是目录
"""
os_path = os.path
#当前位置的名称
base_name = os_path.basename(givenpath)
#上一级路径
dir_name = os_path.dirname(givenpath)
existFlag = os_path.exists(givenpath)
#如果givenpath是以斜杠结尾，就去掉斜杠
if givenpath.endswith("//") :
idx = str(givenpath).rindex("//")
givenpath = str(givenpath)[:idx]
if givenpath.endswith("/") :
idx = str(givenpath).rindex("/")
givenpath = str(givenpath)[:idx]
if base_name.find(".") == -1:
# 如果是目录且不存在，就创建
if not existFlag:
# 创建目录
os.makedirs(givenpath)
return givenpath,dir_type
else:
#如果是文件就检查上一级目录是否创建
flag = os_path.exists(dir_name)
if not flag:
os.makedirs(dir_name)
return givenpath,file_type
def delete_file(self,filePath):
"""
删除指定的HDFS文件。
:param filePath: 文件路径
:return: 1表示删除成功，0删除失败。
"""
try:
check_result = self.__check_hdfs_path(filePath,file_type)
if not check_result:
return {0: "请指定文件路径！"}
except MyHdfsException:
return {0:"文件路径不存在！"}
#true if delete is successful else false.
del_result = self.fs.delete(filePath)
if del_result:
return {1:"删除成功"}
else:
return {0:"删除失败"}
def read_hdfs_file(self,path,lines=100):
"""
读取给定的HDFS文件
:param path: 文件路径
:param lines:需要读取的行数
:return: 1表示成功读取到的字节流序列。0表示读取有问题。
"""
if (not self.fs.exists(path)):
#raise RuntimeError("路径(%s)不存在！" % (path))
return {0:"路径(%s)不存在！" % (path)}
file_status = self.fs.get_file_status(path)
#判断是否是文件
if (file_status.type==file_type):
resp = self.fs.open(path) #urllib3.response.HTTPResponse
#resp.flush()
line_bytes = resp.readline() #bytes
#记录读取的行数
lines = int(lines)
lines -= 1
list_bytes = []
while len(line_bytes)>0 and lines>=0:
#bytes==>str
list_bytes.append(line_bytes.decode("ISO-8859-1").strip())
line_bytes = resp.readline()
lines -= 1
resp.close()
return {1:list_bytes}
else:
#raise RuntimeError("请输入文件路径！")
return {0:"请输入文件路径！"}
def get_subpath(self,pathInfo=PathInfo('/',[]),recursive=False):
"""
（循环）遍历FS中指定路径下的内容：文件或目录
:param path: 封装路径信息的对象
:param recursive: 是否需要循环遍历完给定路径下所有的子路径
:return:路径信息
"""
self_path = pathInfo.name
sub_pathes = pathInfo.pathInfos
if(not self.fs.exists(self_path)):
return "路径(%s)不存在！"%(self_path)
statuses = self.fs.list_status(self_path)
if (not recursive):
# path.pathInfos = fs.listdir(self_path)
if (self_path == "/"):
pathInfo.pathInfos = [self_path+i for i in self.fs.listdir(self_path)]
else:
pathInfo.pathInfos = [self_path +"/" + i for i in self.fs.listdir(self_path)]
else:
for status in statuses:
currentPathSuffix = status.pathSuffix
# 转成详细的路径
if (self_path == "/"):
currentPathSuffix = self_path + currentPathSuffix
else:
currentPathSuffix = self_path + "/" + currentPathSuffix
if (status.type == dir_type):
# 如果是目录，就封装到对象并继续遍历
tmp = PathInfo(currentPathSuffix, [])
sub_pathes.append(tmp)
self.get_subpath(tmp, recursive)
else:
# 如果是文件，就直接返回
sub_pathes.append(currentPathSuffix)
pathInfo.pathInfos = sub_pathes
if __name__ == '__main__':
# result = PathInfo('/', [])
# get_subpath(result,True)
#print(result)
#path = "/user/yarn/mapreduce/mr-framework"
#path = "/testdata"
#hdfspath="/testdata/2013_trip_data_test.csv" #
# #path="/testdata/北京大兴机场.jpg"
# list = [i.decode("utf-8") for i in read_hdfs_file(hdfspath,3)]
# print(list)
#download_to_local(path,"d:/tg/gaohan.csv")
#path="d:/datas/gh/tg"
#print(download_to_local(hdfspath, path))
# myhdfs = MyHdfs(namenodes)
# myhdfs.download_to_local(hdfspath, path)
#app.run()
http_server = WSGIServer(("0.0.0.0",5010),app)
http_server.serve_forever()
