从.net parallel角度解读spark

 1 using System;
 2 using System.Collections.Generic;
 3 using System.Linq;
 4 using System.Text;
 5 using System.Threading.Tasks;
 6 
 7 namespace WordCountDemo
 8 {
 9     using System.IO;
10     using System.Threading;
11     class Program
12     {
13         /// <summary>
14         /// 我们以计算一篇文章中单词的个数为例子
15         /// （计算文章单词个数的demo简直就是各种大数据计算的HelloWorld）。
16         /// 
17         /// WordCountFlow是数单词程序
18         /// WordCountDetail对WordCountFlow函数每一行进行拆解并做了详细解释。
19         /// </summary>
20         /// <param name="args"></param>
21         static void Main(string[] args)
22         {
23             string filePath = @"D:\BigDataSoftware\spark-2.1.0-bin-hadoop2.7\README.md";
24 
25             WordCountFlow(filePath);
26             Console.WriteLine("----------------------");
27             WordCountDetail(filePath);            
28         }
29 
30         /// <summary>
31         /// 数单词的程序流程
32         /// </summary>
33         /// <param name="filePath"></param>
34         static void WordCountFlow(string filePath)
35         {
36             File.ReadAllLines(filePath).AsParallel()
37                 .SelectMany(t => t.Split(‘ ‘))
38                 .Select(t => new { word = t, tag = 1 })
39                 .GroupBy(t => t.word).Select(t => new { word = t.Key, count = t.Select(p => p.tag).Aggregate((a, b) => a + b) })
40                 // 如果对Aggregate函数不熟悉，上面代码等同于下行
41                 //.GroupBy(t => t.word).Select(t => new { word = t.Key, count = t.Sum(p => p.tag) });
42                 .ForAll(t => Console.WriteLine($"ParationId:{Thread.CurrentThread.ManagedThreadId}   ({t.word}-{t.count})"));
43         }
44 
45         /// <summary>
46         /// 数单词程序流程的详细解释
47         /// </summary>
48         /// <param name="filePath"></param>
49         static void WordCountDetail(string filePath)
50         {
51             // 读取整篇文章，文章每一行将作为一个string存储到数组lines
52             string[] lines = File.ReadAllLines(filePath);
53             // AsParallel()是Parallel类库的核心方法，具体的意思是将string[] lines这个数组分割成几个分区（Partition）。
54             // 假设这篇文章有500行，那么这个方法会会把string[500]-lines分解成 (string[120] partitionA), 
55             // (string[180] partitionB), (string[150] partitionC),(...) 等几个Partition
56             // .net runtime将当前程序的负载（主要是cpu使用情况）情况为依据的分区算法来确定到底要分成几个Partition，
57             // 我们可以大概认为cpu有几个逻辑核（不准确），就会被分解成几个Partition。
58             // 后续的计算中.net runtime将会针对每一个partition申请一个单独的线程来处理.
59             // 比如：partitionA由001号线程处理，partitionB由002号线程处理。。。
60             ParallelQuery<string> parallelLines = lines.AsParallel();
61             // linesA,linesB,linesC...数组中存储的每一行根据空格分割成单词，结果仍然是存放在ParallelQuery<string>这种分块的结构中
62             // 下面带有****的注释，如果对函数式编程没有了解，可以直接忽略。
63             // ****如果对函数式编程有所了解，会知道lambda天生lazy的，如果下面这行代码打个断点，当debug到这行代码的时候，
64             // ****鼠标移动到parallelWords上时，我们不会看到每一个单词，
65             // ****runtime并没有真正将每一行分解成单词，这行代码仅仅是一种计算逻辑。
66             ParallelQuery<string> parallelWords = parallelLines.SelectMany(t => t.Split(‘ ‘));
67             // 将每一个单子加上标记1，这行代码返回的类型为ParallelQuery<var>,var为runtime自动判断，此处var的类型的实际应该为 
68             // class 匿名类型
69             // { 
70             //        public word {get;set;}
71             //        public tag {get;set}
72             //}    
73             var wordparis = parallelWords.Select(t => new { word = t, tag = 1 });
74             // 根据单词进行分组，同一个分组中的单词个数求和，类似于如下sql  select word,count(tag) from wordparis group by word
75             // 注意，此处同样的单词可能分布在不同的分区中，比如英语中常见的"the"，可能partitionA中有3个"the",partitionB中有2个“the"，
76             // 但是partitionA和partitionB分别被不同的线程处理，如果runtime足够聪明的话，他应该先计算partitionA的the的个数（the,3）,
77             // 然后计算partitionB的the的个数(the,2),最后将整个partition合并并且重新分割（shuffle），在做后续的计算
78             // shuffle后partition的分区和之前partition里面的数据会不同。
79             // 此处wordcountParis的类型为
80             // class 匿名类型
81             // { 
82             //        public word {get;set;}
83             //        public count {get;set}
84             //}
85             var wordcountParis = wordparis.GroupBy(t => t.word).Select(t => new { word = t.Key, count = t.Select(p => p.tag).Aggregate((a, b) => a + b) });
86             // 打印结果。由于线程执行的乱序，可以看到输出的partitionId也是乱序。
87             wordcountParis.ForAll(t => Console.WriteLine($"ParationId:{Thread.CurrentThread.ManagedThreadId}   ({t.word}-{t.count})"));
88         }
89     }
90 }