海量数据快速查找出现频率topN

小咪咪 2023-06-17 12:58 21阅读 0赞

海量数据寻找出现频率最多的topN，且内存受到限制，该如何处理呢？主要思想就是分治。如果使用框架处理，hadoop的MapReduce运算框架是可以解决的，且是分布式的处理。但是单机版该如何处理呢。空闲时间使用python3实现了一版，做一下记录。

import collections
    import re
    import sys
    import time
    
    """ 找出一个超大文件中出现频率最高的100个词 这里使用分治的方式先将大文件拆分为多个小文件，比如这里将大文件拆分为100个小文件。 需要思考的问题是每个小文件，需要选择前多少位的数据作为下次排序的数据呢？ """
    
    def split4():
        """ 如果是文本文件，可以使用逐行读取。 但是逐行读取的速度是比较慢的。所以内存不是特别小的情况下，可以使用批量读取。速度要快N倍 """
    
        with open("E:/a/english.txt", "r", encoding="UTF-8") as f:
            i = 0
            file_writer = open('E:/a/0.txt', 'a', encoding="UTF-8");
            while True:
                # buff = f.readline()
                # 如果是大文件，需要将这里的缓冲区设置大些，否则文件切分会消耗比较多的时间
                buff = f.read(1024 * 1024)
                if buff:
                    if i < 100:
                        # 先将列表中第一个元素写入文件中
                        file_writer.write(buff);
                        # 关闭当前句柄
                        file_writer.close();
                        # 重新创建一个新的句柄，等待写入下一个切片元素。注意这里文件名的处理技巧。
                        file_writer = open('E:/a/' + str(i) + '.txt', 'a', encoding="UTF-8");
                        i=i+1
                    else:
                        i = 0
                else:
                    break
    
    
    def top100(data,i):
        if i == 100:
            return
        with open("E:/a/"+str(i)+".txt", "r", encoding="UTF-8") as f:
            words_dict = { }
            while True:
                buff = f.read(1024 * 1024)
                if buff:
                    words = re.split('[.,]', buff.replace(" ", ","))
                    for word in words:
                        if word == '' or word is None:
                            continue
                        number = words_dict.get(word)
                        if number == '' or number is None:
                            words_dict[word] = 1
                        else:
                            number = number + 1
                            words_dict[word] = number
                else:
                    break
    
            # 对words_dict进行排序，选择前100频率最高的单词。这里的排序是python3提供的。lambda输入x，输出value
            # 这里返回的是一个tupe集合，而不是dict集合，所以需要注意。
            sorted_list = sorted(words_dict.items(), key=lambda x:x[1], reverse=True)
            index = 0
            for item in sorted_list:
                if index == 100:
                    break
                data.append(item)
                index = index+1
    
        top100(data,i+1)
    
    def merge(data):
        """data是tupe集合，所以需要将再次进行拆解"""
        sorted_words_dict ={ }
        # 这里返回的是一个元祖，key为单词，value为单词在各文件出现的次数
        for item in data:
            value = sorted_words_dict.get(item[0])
            if value == '' or value is None:
                sorted_words_dict[item[0]] = 0
            else:
                value = value + item[1]
                sorted_words_dict[item[0]] = value
    
        sorted_words_list = sorted(sorted_words_dict.items(),key=lambda x:x[1],reverse=True)
        print(sorted_words_list)
    
    
    def main():
        print("切分开始时间:%s" % (time.time()))
        split4()
        print("切分结束时间:%s" % (time.time()))
    
        print("排序开始时间:%s" % (time.time()))
        data = []
        top100(data,0)
        print("排序结束时间:%s" % (time.time()))
    
        print("合并开始时间:%s" % (time.time()))
        merge(data)
        print("合并结束时间:%s" % (time.time()))
    
    
    if __name__ == "__main__":
        main()

测试过2G的英文文档可跑通，展示效果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppbmppbjYwMw_size_16_color_FFFFFF_t_70]  
这里使用的是python提供的排序。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppbmppbjYwMw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191129101841959.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppbmppbjYwMw==,size_16,color_FFFFFF,t_70