Python操作高可用HDFS,实现文件按hash值取余从HDFS到本地

Python操作高可用HDFS,实现文件按hash值取余从HDFS到本地

要实现的功能为3台机器从HDFS同步文件，对文件取hash值对3取余。将不同余数的文件下载到不同机器上。

pip install hdfs
pip install hashlib

import hashlib
from hdfs.client import Client
import time
#hdfs Namenode地址
hdfshost1="http://hadoop102:50070/"
hdfshost2="http://hadoop103:50070/"
#hdfs 源文件路径
hdfspath="/test"
#本地目标路径
localpath="D:\cod"
#对取余数
modulo=3
#取余结果
remainder=1
def download1():
    client = Client(hdfshost2)
    filelist=client.list(hdfs_path=hdfspath,status=True)
    for files in filelist:
        filename=files[0]
        md5 = hashlib.md5()
        md5.update(filename.encode('utf-8'))
        hash_num=int(md5.hexdigest(), 16)
        num=hash_num%modulo
        if(num==remainder):
            print("下载文件"+files[0]+".......")
            client.download(hdfs_path=hdfspath+"/"+files[0],local_path=localpath,overwrite=True)
            print("下载文件" + files[0]+"完成！")
            print("删除文件" + files[0])
            client.delete(hdfs_path=hdfspath+"/"+files[0])
            print("删除文件" + files[0] + "完成！")
            with open('log', 'a') as f:  # 设置文件对象
                f.write("删除文件："+hdfspath+str(files[0])+" 删除时间: "+time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())+"\n")
def download2():
    client = Client(hdfshost1)
    filelist=client.list(hdfs_path=hdfspath,status=True)
    for files in filelist:
        filename=files[0]
        md5 = hashlib.md5()
        md5.update(filename.encode('utf-8'))
        hash_num=int(md5.hexdigest(), 16)
        num=hash_num%modulo
        if(num==remainder):
            print("下载文件"+files[0]+".......")
            client.download(hdfs_path=hdfspath+"/"+files[0],local_path=localpath,overwrite=True)
            print("下载文件" + files[0]+"完成！")
            print("删除文件" + files[0])
            client.delete(hdfs_path=hdfspath+"/"+files[0])
            print("删除文件" + files[0] + "完成！")
            with open('log', 'a') as f:  # 设置文件对象
                f.write("删除文件："+hdfspath+str(files[0])+" 删除时间: "+time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())+"\n")
try:
    download1()
except:
    download2()

因为hash()函数不同进程hash值不同，去选用了hashlib库的MD5进行取hash值。两个hdfshost为两个namenode地址。因为hdfs这个库没法获取namenode状态，所以选用了try: except：做异常处理。 最后会将同步好的脚本进行删除。将删除记录写入log文件。