探索C#之微型MaPReduce

MapReduce近幾年比較熱的分布式計(jì)算編程模型，以C#為例簡單介紹下MapReduce分布式計(jì)算。

閱讀目錄

1. 背景
2. Map實(shí)現(xiàn)
3. Reduce實(shí)現(xiàn)
4. 支持分布式
5. 總結(jié)

背景

某平行世界程序猿小張接到Boss一項(xiàng)任務(wù)，統(tǒng)計(jì)用戶反饋內(nèi)容中的單詞出現(xiàn)次數(shù)，以便分析用戶主要習(xí)慣。文本如下：

        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.";

小張作為藍(lán)翔高材生，很快就實(shí)現(xiàn)了：

   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);            }

作為有上進(jìn)心的青年，小張決心對(duì)算法進(jìn)行抽象封裝，并支持多節(jié)點(diǎn)計(jì)算。小張把這個(gè)統(tǒng)計(jì)次數(shù)程序分成兩個(gè)大步驟：分解和計(jì)算。第一步：先把文本以某維度分解映射成最小獨(dú)立單元。 (段落、單詞、字母維度)。第二部：把最小單元重復(fù)的做合并計(jì)算。小張參考MapReduce論文設(shè)計(jì)Map、Reduce如下：

Map實(shí)現(xiàn)

Mapping

Mapping函數(shù)把文本分解映射key，value形式的最小單元，即<單詞，出現(xiàn)次數(shù)(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

為了減少數(shù)據(jù)通信開銷，mapping出的鍵值對(duì)數(shù)據(jù)在進(jìn)入真正的reduce前，進(jìn)行重復(fù)鍵合并。也相對(duì)于提前進(jìn)行預(yù)計(jì)算一部分，加快總體計(jì)算速度。 輸出格式為(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主要用來分組劃分,把不同節(jié)點(diǎn)的統(tǒng)計(jì)數(shù)據(jù)按照key進(jìn)行分組。其輸出格式為： (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實(shí)現(xiàn)

Reducing函數(shù)接收，分組后的數(shù)據(jù)進(jìn)行最后的統(tǒng)計(jì)計(jì)算。

 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;        }

封裝調(diào)用如下：

 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