C# 實作 MapReduce：LINQ 風格擴充方法與 WordCount 範例

MapReduce 核心概念

• Map：將每筆輸入轉換為 0~N 個 Key-Value Pair
• Shuffle/Group：依 Key 分組，收集所有 Value 成序列
• Reduce：對每組 Key + Values 執行歸約，產出最終結果

泛型擴充方法實作

using System;
using System.Collections.Generic;
using System.Linq;

public static class EnumerableMapReduce
{
    /// 

    /// MapReduce 擴充方法
    /// 
    /// 輸入元素型別
    /// 分組鍵型別
    /// Map 輸出值型別
    /// Reduce 結果型別
    /// 資料來源
    /// Map 函式：單一輸入 → 多組 Key-Value
    /// Reduce 函式：Key + Values 序列 → 結果
    /// Key → Result 字典
    public static Dictionary MapReduce(
        this IEnumerable source,
        Func>> map,
        Func, TResult> reduce)
    {
        // Phase 1: Map + Group (Shuffle)
        var grouped = new Dictionary>();
        
        foreach (var item in source)
        {
            foreach (var kv in map(item))
            {
                if (!grouped.TryGetValue(kv.Key, out var list))
                {
                    list = new List();
                    grouped[kv.Key] = list;
                }
                list.Add(kv.Value);
            }
        }
        
        // Phase 2: Reduce
        var result = new Dictionary();
        foreach (var g in grouped)
        {
            result[g.Key] = reduce(g.Key, g.Value);
        }
        return result;
    }
}

為什麼 Map 回傳 IEnumerable<KeyValuePair> 而非單一值？

因為單一輸入可能產生多組 Key-Value (如單一文檔拆分多個單字)，需用 yield return 逐一產出。Lambda 無法使用 yield，故 Map 需為獨立方法或區塊。

範例：依年齡分組人名 (WordCount 變體)

public class Person
{
    public int Id { get; set; }
    public string Name { get; set; }
    public int Age { get; set; }
}

static void Main()
{
    var people = new List
    {
        new Person { Id = 1, Name = "user1", Age = 23 },
        new Person { Id = 2, Name = "user2", Age = 24 },
        new Person { Id = 3, Name = "user3", Age = 23 },
        new Person { Id = 4, Name = "user4", Age = 25 },
        new Person { Id = 5, Name = "user5", Age = 20 }
    };

    // 只處理年齡 > 22 的人
    // Map: 年齡 → 人名
    // Reduce: 將同齡人名用逗號連接
    var result = people.MapReduce(
        MapByAge,
        (age, names) => string.Join(",", names)
    );

    foreach (var kv in result.OrderBy(x => x.Key))
        Console.WriteLine($"{kv.Key}: {kv.Value}");
}

// Map 方法：使用 yield return 產生多組結果
static IEnumerable> MapByAge(Person p)
{
    if (p.Age > 22)
        yield return new KeyValuePair(p.Age, p.Name);
}

輸出結果

23: user1,user3
24: user2
25: user4

進階：平行化版本 (PLINQ)

public static Dictionary MapReduceParallel(
    this IEnumerable source,
    Func>> map,
    Func, TResult> reduce)
{
    // Parallel Map + Group (使用 ConcurrentDictionary)
    var grouped = new System.Collections.Concurrent.ConcurrentDictionary>();
    
    source.AsParallel().ForAll(item =>
    {
        foreach (var kv in map(item))
        {
            grouped.AddOrUpdate(kv.Key,
                _ => new List { kv.Value },
                (_, list) => { lock (list) list.Add(kv.Value); return list; });
        }
    });
    
    // Reduce (可平行)
    var result = new Dictionary();
    foreach (var g in grouped)
        result[g.Key] = reduce(g.Key, g.Value);
    return result;
}

與 LINQ GroupBy 比較

適用場景

• 日誌分析：Map 擷取 (IP, 1) → Reduce 統計 IP 存取次數
• 文檔處理：Map 拆分單字 → Reduce 詞頻統計
• ETL 轉換：Map 正規化 → Reduce 聚合
• 教學/原型：快速驗證 MapReduce 邏輯再移植至分散式平台

這個擴充方法將 MapReduce 模式帶入 C# LINQ 生態，讓開發者能用熟悉的語法表達分散式計算邏輯，是從單機邁向大數據處理的優秀踏腳石。