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()