基于 Python 实现 MapReduce

from collections import defaultdictdef mapper(word):    return word, 1
def reducer(key_value_pair):    key, values = key_value_pair    return key, sum(values)def map_reduce_function(input_list, mapper, reducer):    '''    - input_list: 字符列表    - mapper: 映射函数，将输入列表中的每个元素映射到一个键值对    - reducer: 聚合函数，将映射结果中的每个键值对聚合到一个键值对    - return: 聚合结果    '''    map_results = map(mapper, input_list)    shuffler = defaultdict(list)    for key, value in map_results:        shuffler[key].append(value)    return map(reducer, shuffler.items())
if __name__ == "__main__":    words = "python best language".split(" ")    result = list(map_reduce_function(words, mapper, reducer))    print(result)

from collections import defaultdictimport threading
class MapReduceThread(threading.Thread):    def __init__(self, input_list, mapper, shuffler):        super(MapReduceThread, self).__init__()        self.input_list = input_list        self.mapper = mapper        self.shuffler = shuffler
    def run(self):        map_results = map(self.mapper, self.input_list)        for key, value in map_results:            self.shuffler[key].append(value)
def reducer(key_value_pair):    key, values = key_value_pair    return key, sum(values)def mapper(word):    return word, 1def map_reduce_function(input_list, num_threads):    shuffler = defaultdict(list)    threads = []    chunk_size = len(input_list) // num_threads        for i in range(0, len(input_list), chunk_size):        chunk = input_list[i:i+chunk_size]        thread = MapReduceThread(chunk, mapper, shuffler)        thread.start()        threads.append(thread)        for thread in threads:        thread.join()
    return map(reducer, shuffler.items())
if __name__ == "__main__":    words = "python is the best language for programming and python is easy to learn".split(" ")    result = list(map_reduce_function(words, num_threads=4))    for i in result:        print(i)

from collections import defaultdictimport multiprocessing
def mapper(chunk):    word_count = defaultdict(int)    for word in chunk.split():        word_count[word] += 1    return word_count
def reducer(word_counts):    result = defaultdict(int)    for word_count in word_counts:        for word, count in word_count.items():            result[word] += count    return result
def chunks(lst, n):    for i in range(0, len(lst), n):        yield lst[i:i + n]
def map_reduce_function(text, num_processes):    chunk_size = (len(text) + num_processes - 1) // num_processes    chunks_list = list(chunks(text, chunk_size))
    with multiprocessing.Pool(processes=num_processes) as pool:        word_counts = pool.map(mapper, chunks_list)
    result = reducer(word_counts)    return result
if __name__ == "__main__":    text = "python is the best language for programming and python is easy to learn"    num_processes = 4    result = map_reduce_function(text, num_processes)    for i in result:        print(i, result[i])

from datetime import datetimeimport multiprocessingdef chunked_file_reader(file_path:str, chunk_size:int):    """    生成器函数：分块读取文件内容    - file_path: 文件路径    - chunk_size: 块大小,默认为1MB    """    with open(file_path, 'r', encoding='utf-8') as file:        while True:            chunk = file.read(chunk_size)            if not chunk:                break            yield chunk
def search_in_chunk(chunk:str, keyword:str):    """在文件块中搜索关键字    - chunk: 文件块    - keyword: 要搜索的关键字    """    lines = chunk.split('\n')    for line in lines:        if keyword in line:            print(f"找到了:", line)
def search_in_file(file_path:str, keyword:str, chunk_size=1024*1024):    """在文件中搜索关键字    file_path: 文件路径    keyword: 要搜索的关键字    chunk_size: 文件块大小,为1MB    """    with multiprocessing.Pool() as pool:        for chunk in chunked_file_reader(file_path, chunk_size):            pool.apply_async(search_in_chunk, args=(chunk, keyword))        if __name__ == "__main__":    start = datetime.now()    file_path = "file.txt"    keyword = "张三"    search_in_file(file_path, keyword)    end = datetime.now()    print(f"搜索完成，耗时 {end - start}")