MapReduce分布式计算 入门级实例代码

【实例简介】
【实例截图】
【核心代码】

namespace mapreduce
{
    public class Master<T>
    {
        public Dictionary<T, int> Result = new Dictionary<T, int>();
        public  void Merge(Dictionary<T, int> list)
        {
            foreach (var item in list)
            {
                lock (this)
                {
                    if (Result.ContainsKey(item.Key))
                        Result[item.Key] = item.Value;
                    else
                        Result.Add(item.Key, item.Value);
                }
            }
        }
    }
    public class MicroMapReduce<T>
    {
        private Master<T> master { get; set; }
        public MicroMapReduce(Master<T> m)
        {
            master = m;
        }
        public Dictionary<T, int> DMapReduce(IEnumerable<T> list)
        {
           var p= Partition(list);
           foreach (var item in p)
           {
               WorkerNode(item);
           }
           Thread.Sleep(1000);  //模拟完成通知

           return master.Result;
        }
        /// <summary>
        /// 单机版
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="list"></param>
        /// <returns></returns>
        public  Dictionary<T, int> MapReduce(IEnumerable<T> list)
        {
            var map = Map(list);
            var reduce = Reduce(map);
            return reduce;
        }
       
        public List<IEnumerable<T>> Partition(IEnumerable<T> list)
        {
            var temp =new List<IEnumerable<T>>();
            temp.Add(list);
            temp.Add(list);
            return temp;
        }
        public void WorkerNode(IEnumerable<T> list)
        {
            new Thread(() =>
            {
                var map = Map(list);
                var reduce = Reduce(map);
                master.Merge(reduce);
            }).Start();
        }
        public IEnumerable<Group<T, int>> Map(IEnumerable<T> list)
        {
            var step1 = Mapping(list);
            var step2 = Combine(step1);
            var step3 = Partitioner(step2);
            return step3;
        }
        public IEnumerable<Tuple<T, int>> Mapping(IEnumerable<T> list)
        {
            foreach (T sourceVal in list)
                yield return Tuple.Create(sourceVal, 1);
        }
        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;
        }

        public Dictionary<T, int> Reduce(IEnumerable<Group<T, int>> groups)
        {
            var step1 = Reducing(groups);
            return step1;
        }
        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;
        }
        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 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;
            }
        }
    }
}