Python MapReduce 深度解析與效能最佳化實踐

import multiprocessing as mp
from collections import defaultdict
import itertools

def mapper(data):
    words = data.split()
    results = []
    for word in words:
        results.append((word.lower(), 1))
    return results

def combiner(mapped_data):
    combined = defaultdict(int)
    for key, count in mapped_data:
        combined[key] += count
    return list(combined.items())

def reducer(partitioned_data):
    reduced = {}
    for key, group in itertools.groupby(sorted(partitioned_data), key=lambda x: x[0]):
        reduced[key] = sum(item[1] for item in group)
    return reduced

def partition(data, num_workers):
    partitions = [[] for _ in range(num_workers)]
    for idx, pair in enumerate(data):
        partitions[idx % num_workers].append(pair)
    return partitions

if __name__ == '__main__':
    corpus = "Advanced parallel processing in Python requires mastering MapReduce"
    chunks = [corpus] * 4  # 模擬 4 個資料區塊
    with mp.Pool(processes=4) as pool:
        mapped = pool.map(mapper, chunks)
        combined = pool.map(combiner, mapped)
        flat_data = [pair for sublist in combined for pair in sublist]
        partitions = partition(flat_data, 4)
        reduced_results = pool.map(reducer, partitions)
    
    final_result = defaultdict(int)
    for partial in reduced_results:
        for key, count in partial.items():
            final_result[key] += count
    print(dict(final_result))

def stream_mapper(iterator):
    for line in iterator:
        for word in line.strip().split():
            yield (word.lower(), 1)

def stream_reducer(mapped_stream):
    current_word = None
    current_count = 0
    for key, count in mapped_stream:
        if key != current_word:
            if current_word is not None:
                yield (current_word, current_count)
            current_word = key
            current_count = count
        else:
            current_count += count
    if current_word is not None:
        yield (current_word, current_count)

def sorted_stream_stream(input_stream):
    return sorted(input_stream, key=lambda x: x[0])

if __name__ == '__main__':
    lines = iter([
        "MapReduce in Python is powerful",
        "Python can leverage MapReduce for big data",
        "Data processing using MapReduce is scalable"
    ])
    mapped_stream = stream_mapper(lines)
    sorted_mapped = sorted_stream_stream(mapped_stream)
    reduced_stream = stream_reducer(sorted_mapped)
    for word_count in reduced_stream:
        print(word_count)

from pyspark.sql import SparkSession
import numpy as np

spark = SparkSession.builder.appName("MapReduceOptimization").getOrCreate()
data = spark.sparkContext.parallelize(range(1000000), 8)

def process_partition(partition):
    data_partition = np.array(list(partition))
    # 套用向量化操作以提升效能
    transformed = data_partition * 2  # 示例操作
    return iter(transformed.tolist())

result = data.mapPartitions(process_partition).collect()
spark.stop()

from queue import Queue
from threading import Thread
import time

def producer(queue, data):
    for item in data:
        # 模擬資料提取延遲
        time.sleep(0.01)
        queue.put(item)
    queue.put(None)  # 哨兵值表示完成

def processor(in_queue, out_queue):
    while True:
        item = in_queue.get()
        if item is None:
            out_queue.put(None)
            break
        # 模擬轉換延遲
        time.sleep(0.02)
        processed = item * 2
        out_queue.put(processed)

def consumer(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        # 處理或儲存已處理的專案
        print(item)

if __name__ == '__main__':
    data = range(10)
    q1 = Queue()
    q2 = Queue()

    p = Thread(target=producer, args=(q1, data))
    pr = Thread(target=processor, args=(q1, q2))
    c = Thread(target=consumer, args=(q2,))

    p.start()
    pr.start()
    c.start()

    p.join()
    pr.join()
    c.join()