要实现的功能为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文件。
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)