阅读量:4
数据
数据预处理
df = pd.read_csv(file_path, encoding='ANSI')
csv的编码方式一定要用 ANSI。要不然会出现各种报错
import pandas as pd from datetime import datetime # 读取CSV文件 file_path = 'book_douban.csv' df = pd.read_csv(file_path, encoding='ANSI') # 定义一个函数来提取年份 def extract_year(publish_date): publish_date= str(publish_date).strip() try: # 尝试不同的日期格式 for fmt in ('%Y/%m', '%Y/%m/%d', '%Y年%m月', '%Y'): try: return int(datetime.strptime(publish_date, fmt).year) except ValueError: pass except TypeError: return None # 应用函数提取年份 df['出版时间'] = df['出版时间'].apply(extract_year) # 清洗数据，删除评分空值 df = df[df['评分'] != 0] df = df[df['书名'] != '不存在未出版的错误条目'] df = df[df['书名'] != '点击上传封面图片'] # 左右两边去除空格 for column in df.columns: if column == "书名" or column == "出版社" or column == "作者": df[column] = df[column].str.strip() # 去除重复行。除了第一列id不一样，其他均一样 (806和807行) # df = df.drop_duplicates(keep=False) df = df.drop_duplicates(subset=df.columns[1:], keep='first') # 将清洗后的数据写入新的CSV文件 output_file_path = 'cleaned_books.csv' df.to_csv(output_file_path, index=False, encoding='ANSI') print(f"数据清洗完成,并写入新文件：{output_file_path}") 布隆过滤器
由于相同的书籍可能出于多个出版社，为避免书籍评分重复导入，实现BF过滤器，每当有一部书籍被存储后将其加入BF过滤器，并能够使用BF过滤器查询上述书籍是否已经被存储。
import hashlib import math import mmh3 # 使用 mmh3 库作为哈希函数 import random class SimpleBloomFilter: def __init__(self, items_count, fp_prob): # 计算所需的位数组的大小和哈希函数的数量 self.size = self.get_size(items_count, fp_prob) self.hash_count = self.get_hash_count(self.size, items_count) self.bit_array = [0] * self.size def add(self, item): # 添加元素到布隆过滤器 digests = [] for i in range(self.hash_count): # 使用不同的种子生成不同的哈希值 digest = mmh3.hash(item, i) % self.size digests.append(digest) self.bit_array[digest] = 1 def check(self, item): # 检查元素是否可能存在于布隆过滤器 for i in range(self.hash_count): digest = mmh3.hash(item, i) % self.size if self.bit_array[digest] == 0: # 如果有一个位是0，则元素肯定不在集合中 return False return True @classmethod def get_size(self, n, p): # 计算位数组的大小 m = -(n * math.log(p)) / (math.log(2) ** 2) return int(m) @classmethod def get_hash_count(self, m, n): # 计算哈希函数的数量 k = (m / n) * math.log(2) return int(k) 布隆过滤器原理
https://blog.csdn.net/qq_41125219/article/details/119982158
布隆过滤器实际上是一个很长的二进制向量和一系列随机映射函数。可以用于解决Redis缓存穿透问题
Redis中的布隆过滤器底层是一个大型位数组（二进制数组）+多个无偏hash函数。无偏hash函数就是能把元素的hash值计算的比较均匀的hash函数，能使得计算后的元素下标比较均匀的映射到位数组中。
如下就是一个简单的布隆过滤器示意图，其中k1、k2代表增加的元素，a、b、c即为无偏hash函数，最下层则为二进制数组。
在布隆过滤器增加元素之前，首先需要初始化布隆过滤器的空间，也就是上面说的二进制数组，除此之外还需要计算无偏hash函数的个数。布隆过滤器提供了两个参数，分别是预计加入元素的大小n，运行的错误率f。布隆过滤器中有算法根据这两个参数会计算出二进制数组的大小l，以及无偏hash函数的个数k。
添加元素
往布隆过滤器增加元素，添加的key需要根据k个无偏hash函数计算得到多个hash值，然后对数组长度进行取模得到数组下标的位置，然后将对应数组下标的位置的值置为1
- 通过k个无偏hash函数计算得到k个hash值
- 依次取模数组长度，得到数组索引
- 将计算得到的数组索引下标位置数据修改为1
查询元素
- 通过k个无偏hash函数计算得到k个hash值
- 依次取模数组长度，得到数组索引
- 判断索引处的值是否全部为1，如果全部为1则存在（这种存在可能是误判），如果存在一个0则必定不存在
误报是由于hash冲突
布隆过滤器的缺点：
- 有点一定的误判率，但是可以通过调整参数来降低
- 无法获取元素本身
- 很难删除元素
数据写入mongodb以及数据合并
mongodb下载
https://www.mongodb.com/try/download/community
mongodb客户端工具MongoDB Compass下载
https://www.mongodb.com/try/download/compass
import pymongo import pandas as pd from pybloom_live import BloomFilter import re # 连接MongoDB from SimpleBloomFilter import SimpleBloomFilter client = pymongo.MongoClient("mongodb://localhost:27017/") db = client["book_database"] collection = db["books"] # 使用布隆过滤器 items_count = 60000 # 预计存储的元素数量 fp_prob = 0.001 # 可接受的错误率 bf = SimpleBloomFilter(items_count, fp_prob) # 读取CSV文件 file_path = 'cleaned_books.csv' df = pd.read_csv(file_path, encoding='ANSI') # 准备数据，准备存储到MongoDB data_to_store = df.to_dict(orient='records') # 这个作者数据太脏了，是一个作者，但是名字写法不同 def is_same_author(name1, name2): # 去除字符串中的括号和特殊字符 name1 = re.sub(r'\[.*?\]', '', name1) # 去除类似 [法] 这样的内容 name2 = re.sub(r'\[.*?\]', '', name2) name1 = re.sub(r'\(.*?\)', '', name1) name2 = re.sub(r'\(.*?\)', '', name2) name1 = re.sub(r'（.*?）', '', name1) # 去除类似 [法] 这样的内容 name2 = re.sub(r'（.*?）', '', name2) # 将字符串中的英文和数字转换为小写 name1 = re.sub(r'[A-Za-z0-9]+', lambda x: x.group().lower(), name1) name2 = re.sub(r'[A-Za-z0-9]+', lambda x: x.group().lower(), name2) # 对中文进行分词（这里简化处理，实际情况可能需要更复杂的分词算法） # 假设我们只取中文名字的第一部分作为代表 name1_chinese = re.search(r'[\u4e00-\u9fa5]+', name1) name2_chinese = re.search(r'[\u4e00-\u9fa5]+', name2) # 如果两个名字都有中文部分，比较中文部分 if name1_chinese and name2_chinese: name1_standard = name1_chinese.group() name2_standard = name2_chinese.group() else: # 如果没有中文部分，比较整个字符串（已去除英文大小写和括号的影响） name1_standard = name1 name2_standard = name2 # 比较两个标准化后的字符串 return name1_standard == name2_standard for book in data_to_store: # 检查布隆过滤器，确定这本书是否已经存储 # 以 '书名+作者' 为唯一标识。 有的书名一样，但是作者不一样。 比如 “系统神学” (但是有的书作者为空，不好之前将书名和作者拼接放到布隆过滤器中去) if bf.check(book['书名']): query = {'书名': book['书名']} search_result = collection.find(query) # 打印查询结果 for _book in search_result: if pd.isna(book['作者']) or pd.isna(_book['作者']) or is_same_author(book['作者'], _book['作者']): if isinstance(_book.get('出版社', ''), str) and pd.isna(_book.get('出版社', '')): existing_publishers = book['出版社'] else: if isinstance(_book.get('出版社', ''), str) and _book.get('出版社', '') != book['出版社']: existing_publishers = [_book.get('出版社', ''), book['出版社']] elif isinstance(_book.get('出版社', ''), list): # 如果已经是列表，则直接使用 existing_publishers = _book.get('出版社', []) if book['出版社'] not in existing_publishers: # print(_book.get('出版社', ''), book['书名']) existing_publishers.append(book['出版社']) existing_publish_time = _book.get('出版时间', '') if not pd.isna(book['出版时间']): if isinstance(_book.get('出版时间', ''), float) and _book.get('出版时间', '') != book['出版时间']: if not pd.isna(_book.get('出版时间', '')): existing_publish_time = [_book.get('出版时间', ''), book['出版时间']] else: existing_publish_time = book['出版时间'] elif isinstance(_book.get('出版时间', ''), list): # 如果已经是列表，则直接使用 existing_publish_time = _book.get('出版时间', []) if book['出版时间'] not in existing_publish_time: existing_publish_time.append(book['出版时间']) new_rating = (float(_book.get('评分', '')) * float(_book.get('评论数量', '')) + float(book['评分']) * float( book['评论数量'])) / (float(_book.get('评论数量', '')) + float(book['评论数量'])) print(existing_publishers,existing_publish_time,book['书名']) # 如果书已存在，更新出版社列表、评分和评论数量 update_result = collection.update_one( {'书名': book['书名']}, { "$set": { "出版社": existing_publishers, "出版时间": existing_publish_time, "评分": new_rating # 确保 new_rating 是计算后的新评分值 }, "$inc": { "评论数量": book['评论数量'] } }, upsert=False ) if update_result.modified_count == 0: # 如果没有修改，说明publishers已经包含在数组中，可以进行去重操作 pass # 只能有一个 break else: # 如果书不存在，添加到数据库和布隆过滤器 collection.insert_one(book) bf.add(book['书名']) # 关闭MongoDB连接 client.close() 数据添加索引
import csv import pandas as pd class BTreeNode: def __init__(self, is_leaf=False): self.keys = [] self.children = [] self.is_leaf = is_leaf class BTree: def __init__(self, t): self.root = BTreeNode(is_leaf=True) self.t = t # Minimum degree def insert(self, publication_time, rating_count, book_info): root = self.root if len(root.keys) == (2 * self.t) - 1: new_root = BTreeNode() self.root = new_root new_root.children.append(root) self.split_child(new_root, 0) self.insert_non_full(self.root, publication_time, rating_count, book_info) def insert_non_full(self, node, publication_time, rating_count, book_info): i = len(node.keys) - 1 if node.is_leaf: node.keys.append((publication_time, rating_count, book_info)) while i >= 0 and publication_time < node.keys[i][0]: node.keys[i + 1] = node.keys[i] i -= 1 node.keys[i + 1] = (publication_time, rating_count, book_info) else: while i >= 0 and publication_time < node.keys[i][0]: i -= 1 i += 1 if len(node.children[i].keys) == (2 * self.t) - 1: self.split_child(node, i) if publication_time > node.keys[i][0]: i += 1 self.insert_non_full(node.children[i], publication_time, rating_count, book_info) def split_child(self, parent, index): child = parent.children[index] new_child = BTreeNode(is_leaf=child.is_leaf) parent.keys.insert(index, child.keys[self.t - 1]) parent.children.insert(index + 1, new_child) new_child.keys = child.keys[self.t:] child.keys = child.keys[:self.t - 1] if not child.is_leaf: new_child.children = child.children[self.t:] child.children = child.children[:self.t] def range_query(self, start_time, end_time, min_rating_count): result = [] self.range_query_helper(self.root, start_time, end_time, min_rating_count, result) return result def range_query_helper(self, node, start_time, end_time, min_rating_count, result): if node.is_leaf: for key in node.keys: if start_time <= key[0] <= end_time and key[1] > min_rating_count: result.append(key) return for i in range(len(node.keys)): if start_time <= node.keys[i][0]: self.range_query_helper(node.children[i], start_time, end_time, min_rating_count, result) self.range_query_helper(node.children[-1], start_time, end_time, min_rating_count, result) def read_csv_and_build_index(file_path, btree, t): with open(file_path, 'r',encoding='ANSI') as file: csv_reader = csv.reader(file) next(csv_reader) # 跳过表头 for row in csv_reader: if row[4] and row[6] and not pd.isnull(row[4]) and not pd.isna(row[6]): publication_time = int(float(row[4])) rating_count = float(row[6]) book_info = row # 保存整行信息 btree.insert(publication_time, rating_count, book_info) # 测试代码 btree = BTree(2) # 设置 t 的值 file_path = 'cleaned_books.csv' # 替换为实际的文件路径 read_csv_and_build_index(file_path, btree, 2) # start_time = 2000 # end_time = 2015 # min_rating_count = 50 start_time = int(input("请输入出版起始时间：")) end_time = int(input("请输入出版终止时间：")) min_rating_count = int(input("请输入最低评论数：")) query_result = btree.range_query(start_time, end_time, min_rating_count) for item in query_result: print(" ".join(item[2])) 生成环境配置文件requirements.txt
导出整个环境的安装包
pip freeze > requirements.txt 导出单个项目安装包
pip install pipreqs pipreqs . 安装依赖
pip install -r requirements.txt -i https://pypi.douban.com/simple 