项目中在收集数据的时候，有时候会遇到日志中存在大量的重复记录，数据在入库之前需要先对日志文件做一个处理，滤掉重复的事件。
执行 pip install pybloom-live即可安装。
02 处理逻辑
1.原始日志数据实时写入文件，每小时生成一个日志文件，文件名为"月日小时.log"，如 “042401.log”(时间和系统时间相差8小时)，原始日志文件路径：/data/logs
2.每小时处理前一小时的原始日志文件，将去重后的日志保存在目标路径下，如"/data/dblog/042101.log"
3.考虑到服务器硬盘空间，需要删除历史数据和处理后的数据文件，服务器中只保留最近24小时的数据
实现代码：
# -*- coding: utf-8 -*-from pybloom_live import BloomFilterimport osimport hashlibimport timeimport datetime#去重class BloomCheckFunction(object): def __init__(self): self.filename = 'bloomFilter.blm' is_exist = os.path.exists(self.filename) #判断文件是否存在 if is_exist: self.bf = BloomFilter.fromfile(open(self.filename, 'rb')) #存在直接打开 储存在内存中 else: self.bf = BloomFilter(100000000, 0.001) #新建一个 储存在内存中 def process_item(self, data): data_encode_md5 = hashlib.md5(data.encode(encoding='utf-8')).hexdigest() if data_encode_md5 in self.bf: # 内容没有更新 丢弃item return False return False else: self.bf.add(data_encode_md5) #内容不存在，新来的 return True return True def save_bloom_file(self): self.bf.tofile(open(self.filename, 'wb'))#获取文件时间def get_filename(): h=datetime.datetime.now()-datetime.timedelta(hours=9) filename=h.strftime('%m%d%H')+".log" logname="process_"+h.strftime('%m%d')+".log" return filename,logname#删除历史文件def delFile(): h=datetime.datetime.now()-datetime.timedelta(hours=33)#可根据硬盘空间大小，确定需要保留多久的数据，这里保留最近24小时的日志文件 del_srcfile="/data/logs/"+h.strftime('%m%d%H')+".log" del_dbfile="/data/dblog/"+h.strftime('%m%d%H')+".log" #print(del_srcfile) if os.path.exists(del_dbfile): os.remove(del_dbfile) if os.path.exists(del_srcfile): os.remove(del_srcfile)def process_data(srcfile,dstfile): bf = BloomCheckFunction()#对象初始化 只需要初始化一遍 with open(dstfile,'w') as fout: with open(srcfile,'r') as fin: for line in fin: if bf.process_item(line): fout.write(line) else: passif __name__=="__main__": start_time=time.time() filename,logname=get_filename() srcfile="/data/logs/"+filename dstfile="/data/dblog/"+filename if os.path.exists(srcfile): try: delFile() process_data(srcfile,dstfile) spend_time=time.time()-start_time with open(logname,"a+") as fin: fin.write("file "+srcfile+" process success , spend times: %.3f s\n" % spend_time) except: with open(logname,"a+") as fin: fin.write("file "+srcfile+" process failed") else: with open(logname,"a+") as fin: fin.write("file "+srcfile+" not exist !!!")
如果本文对你有帮助，欢迎“点赞”、“分享”、“推荐”，也欢迎在留言区互动，感谢！
点关注，一起成长！深耕技术是提升收入的底气，打理资产是守住收入的底气。我的小号【小韭菜的慢富笔记】，专注打工人稳健攒股收息、价值投资实战，不追热点、不玩投机，只分享普通人可复制的慢富思路。感兴趣的朋友，欢迎关注交流，一起慢慢变富~
10个月宝宝每天需要喝多少奶粉？
