探索C#之微型MapReduce

探索C#之微型MaPReduce

MapReduce近几年比较热的分布式计算编程模型，以C#为例简单介绍下MapReduce分布式计算。

阅读目录

1. 背景
2. Map实现
3. Reduce实现
4. 支持分布式
5. 总结

背景

某平行世界程序猿小张接到Boss一项任务，统计用户反馈内容中的单词出现次数，以便分析用户主要习惯。文本如下：

        const string hamlet = @"Though yet of Hamlet our dear brother's deathThe memory be green, and that it us befittedTo bear our hearts in grief and our whole kingdomTo be contracted in one brow of woe,Yet so far hath discretion fought with natureThat we with wisest sorrow think on him,Together with remembrance of ourselves.Therefore our sometime sister, now our queen,The imperial jointress to this warlike state,Have we, as 'twere with a defeated joy,--With an auspicious and a dropping eye,With mirth in funeral and with dirge in marriage,In equal scale weighing delight and dole,--Taken to wife: nor have we herein barr'dYour better wisdoms, which have freely goneWith this affair along. For all, our thanks.Now follows, that you know, young Fortinbras,Holding a weak supposal of our worth,Or thinking by our late dear brother's deathOur state to be disjoint and out of frame,Colleagued with the dream of his advantage,He hath not fail'd to pester us with message,Importing the surrender of those landsLost by his father, with all bonds of law,To our most valiant brother. So much for him.Now for ourself and for this time of meeting:Thus much the business is: we have here writTo Norway, uncle of young Fortinbras,--Who, impotent and bed-rid, scarcely hearsOf this his nephew's purpose,--to suppressHis further gait herein; in that the levies,The lists and full proportions, are all madeOut of his subject: and we here dispatchYou, good Cornelius, and you, Voltimand,For bearers of this greeting to old Norway;Giving to you no further personal powerTo business with the king, more than the scopeOf these delated articles allow.Farewell, and let your haste commend your duty.";

小张作为蓝翔高材生，很快就实现了：

   var content = hamlet.Split(new[] { " ", Environment.NewLine }, StringSplitOptions.RemoveEmptyEntries);            var Wordcount=new Dictionary<string,int>();            foreach (var item in content)            {                if (wordcount.ContainsKey(item))                    wordcount[item] += 1;                else                    wordcount.Add(item, 1);            }

作为有上进心的青年，小张决心对算法进行抽象封装，并支持多节点计算。小张把这个统计次数程序分成两个大步骤：分解和计算。第一步：先把文本以某维度分解映射成最小独立单元。 (段落、单词、字母维度)。第二部：把最小单元重复的做合并计算。小张参考MapReduce论文设计Map、Reduce如下：

Map实现

Mapping

Mapping函数把文本分解映射key，value形式的最小单元，即<单词，出现次数(1)>、<word,1>。

    public IEnumerable<Tuple<T, int>> Mapping(IEnumerable<T> list)        {            foreach (T sourceVal in list)                yield return Tuple.Create(sourceVal, 1);        }

使用，输出为(brow, 1), (brow, 1), (sorrow, 1), (sorrow, 1):

            var spit = hamlet.Split(new[] { " ", Environment.NewLine }, StringSplitOptions.RemoveEmptyEntries);            var mp = new MicroMapReduce<string>(new Master<string>());            var result= mp.Mapping(spit);

Combine

为了减少数据通信开销，mapping出的键值对数据在进入真正的reduce前，进行重复键合并。也相对于提前进行预计算一部分，加快总体计算速度。 输出格式为(brow, 2), (sorrow, 2):

 public Dictionary<T, int> Combine(IEnumerable<Tuple<T, int>> list)        {            Dictionary<T, int> dt = new Dictionary<T, int>();            foreach (var val in list)            {                if (dt.ContainsKey(val.Item1))                    dt[val.Item1] += val.Item2;                else                    dt.Add(val.Item1, val.Item2);            }            return dt;        }

Partitioner

Partitioner主要用来分组划分,把不同节点的统计数据按照key进行分组。其输出格式为： (brow, {(brow,2)},(brow,3)), (sorrow, {(sorrow,10)},(brow,11)):

public IEnumerable<Group<T, int>> Partitioner(Dictionary<T, int> list)        {            var dict = new Dictionary<T, Group<T, int>>();            foreach (var val in list)            {                if (!dict.ContainsKey(val.Key))                    dict[val.Key] = new Group<T, int>(val.Key);                dict[val.Key].Values.Add(val.Value);            }            return dict.Values;        }

Group定义:

    public class Group<TKey, TValue> : Tuple<TKey, List<TValue>>    {        public Group(TKey key)            : base(key, new List<TValue>())        {        }        public TKey Key        {            get            {                return base.Item1;            }        }        public List<TValue> Values        {            get            {                return base.Item2;            }        }    }

Reduce实现

Reducing函数接收，分组后的数据进行最后的统计计算。

 public Dictionary<T, int> Reducing(IEnumerable<Group<T, int>> groups)        {            Dictionary<T, int> result=new Dictionary<T, int>();            foreach (var sourceVal in groups)            {                result.Add(sourceVal.Key, sourceVal.Values.Sum());            }            return result;        }

封装调用如下：

 public IEnumerable<Group<T, int>> Map(IEnumerable<T> list)        {            var step1 = Mapping(list);            var step2 = Combine(step1);            var step3 = Partitioner(step2);            return step3;        }  public Dictionary<T, int> Reduce(IEnumerable<Group<T, int>> groups)        {            var step1 = Reducing(groups);            return step1;        }

  public  Dictionar