1、前一版本算法分析
前一版本的算法主要时间是消耗在合并有交集的类别上,代码结构如:
for i, cls in enumerate(clses):
for cls_j in clses[i+1:]:
# ......
显然这里是一个两层循环,平方的时间复杂度,如果列表有1万个元素,这就变成了1亿次循环(平方就是这么可怕)!
那这里优化的方向:
开始时,只要是从这两方面进行考虑。第二点实现不了,因为循环时对前一个状态有依赖,没法并行,那就只剩下第一点了。减少列表的输入长度是可行的,不过这只是量变,产生不了质变,只能缓解一下问题,可能从耗时可以从二十几秒降到十几秒,只要数据稍微增加一点,时间马上就回去了。
2、相似文章的特点
我们再回头理解理解相似文章,前面我们已经假设将两个文章的simhash码均分为4段之后,如果它们是相似的,那至少有一段是完全相同的。
一段simhash码的子串有16个二进制位,该子串的表达空间最大有2的16次方(共65536),就算有一亿篇文章,如果均分成65536份,那也变得很小(当然实际的分布并不是均匀的),如果从这点出发,那计算的时间复杂度将会大大降低。
重新梳理一下这算法流程,假设有三篇文章,每篇文章有4个simhash子串:
文章1: A1, B1, C1, D1
文章2: A2, B2, C2, D3
文章3: A3, B3, C3, D3
按照我们的假设,如果文章1和文章2相似,则下面4个条件必须至少有一个成立:
A1 == A2
B1 == B2
C1 == C2
D1 == D2
也就是说,也就是只要通过文章1的4个simhash子串就能找到该文章的所有相似文章(当然召回率不会是100%,但是可以忽略,本身simhash就会有误差)。而且,这恰好能满足我们的需求。
那就三个步骤:
很清晰。
3、第四个优化版本
建立simhash子串的索引:
# 建立索引
print(len(data))
start = time.time()
child_key_dict = {}
for item in data:
simhash, article_id = item['simhash'], item['id']
simhash = '0' * (64-len(simhash)) + simhash
item['simhash'] = simhash # 格式simhash
# 不足64位的前面补0,文章id放到第一个值
item = article_id + simhash
for i in range(4):
child_key_dict.setdefault(str(i)+simhash[i*16:i*16+16], []).append(item)
# 文章id到hash值的索引
aid_simhash_dict = {item['id']: item['simhash'] for item in data}
print('time: ', time.time()-start, len(child_key_dict))
筛选候选集及从候选集中过滤出真正的相似文章:
# 循环找相似文章
sim_articles = []
checked_aid_dict = {item['id']: False for item in data}
for item in data:
if checked_aid_dict[item['id']]:
continue
simhash, article_id = item['simhash'], item['id']
checked_aid_dict[article_id] = True
# 可能相似的文章
sim_articles_maybe = []
for i in range(4):
key = str(i) + simhash[i*16:i*16+16]
sim_articles_maybe += child_key_dict[key]
# 去重文章及已经被归类的文章
sim_articles_maybe = set(sim_articles_maybe)
sim_articles_maybe.remove(article_id+simhash)
sim_articles_maybe = list(sim_articles_maybe)
sim_articles_maybe = [aid for aid in sim_articles_maybe if not checked_aid_dict[aid[:-64]]]
if len(sim_articles_maybe) == 0:
sim_articles.append([article_id])
continue
# 找相似的文章
sim_articles_maybe = [[aid[:-64]] + list(aid[-64:]) for aid in sim_articles_maybe]
sim_articles_maybe = np.array(sim_articles_maybe)
sim_aids = find_sim_aids(np.array([article_id]+list(simhash)), sim_articles_maybe)
if len(sim_aids) == 0:
sim_articles.append([article_id])
continue
sim_articles.append([article_id]+sim_aids)
for aid in sim_aids:
checked_aid_dict[aid] = True
print('cluster time: ', time.time()-start, len(sim_articles), len(data))
这基本就是线性的时间复杂度,相对于第三个版本的平方时间复杂度,那是快多了,1.7万的文章进行相似性过滤时间能降到一秒以下。
软烟罗
4、小结
通过索引快速过滤出候选集(有点类似ES的意思),牺牲一点点精度,这就能大大减少无意义的计算。