幻灯片1-SOURCEFORGE.ppt

Team Of Programmers / Parallel Programming Computing 2.0 By Barry.Cswords Team Of Programmers / 并行计算的技术现状 集群和 分布式技术术 虚拟拟化Map-Reduce 分布式 散列存储储 演员员模型 和函数编编程 解决方案网格计算绿色计算云计算云计算? 设计模式 (关联底层设 计) 虚拟资源池 面向数据结构ER模型FP 典型方案负载均衡集群Hadoop CoreHBase，HDFSErlang虚拟机 面向资源事务、存储计算、存储计算存储计算 资源利用率 系统可扩展性 高并发 高吞吐 海量存储 复杂计算 (关联底层设 计) 缺陷 设计 模式顶层 化 业务 关联性强 设计 模式顶层 化 人工干预程度高 设计 模式底层化 代码复杂 设计 模式底层化 PIG等还不成熟 设计 模式底层化 代码晦涩 Team Of Programmers / LISP对列表的处理 先熟悉一下LISP语法 MAP方法 FOLD方法 Scheme（LISP的一种方言）用递归实现，但实际上都可以并行。 (define factorial (lambda (n) (if (= n 1) 1 (* n (factorial (- n 1) (factorial 6) 720 (map (lambda (x) (* x x) (1 2 3 4 5) (1 4 9 16 25) (fold + 0 (1 2 3 4 5) 15 (fold * 1 (1 2 3 4 5) 120 Team Of Programmers / Map-Reduce (Hadoop) package org.myorg; import java.io.IOException; import java.util.*; import org.apache.hadoop.fs.Path; import org.apache.hadoop.conf.*; import org.apache.hadoop.io.*; import org.apache.hadoop.mapred.*; import org.apache.hadoop.util.*; public class WordCount public static void main(String args) throws Exception /以这个类建立任务并起名 JobConf conf = new JobConf(WordCount.class); conf.setJobName(“wordcount“); /输出列表类型规约 conf.setOutputKeyClass(Text.class); conf.setOutputValueClass(IntWritable.class); /Map-Reduce算法类规约 conf.setMapperClass(Map.class); conf.setCombinerClass(Reduce.class); conf.setReducerClass(Reduce.class); /输入输出类型规约 conf.setInputFormat(TextInputFormat.class); conf.setOutputFormat(TextOutputFormat.class); /设置输入、输出路径 FileInputFormat.setInputPaths(conf, new Path(args0); FileOutputFormat.setOutputPath(conf, new Path(args1); JobClient.runJob(conf); （统计计算一篇文章中不同词的个数） C+代码略简练 使用管道上下文实现 Map和Reduce关联 使用工厂实现IO和调用 Team Of Programmers / Map-Reduce (Hadoop)续 public static class Map extends MapReduceBase implements Mapper private final static IntWritable one = new IntWritable(1);/封装个数 private Text word = new Text();/暂存每个词用 public void map(LongWritable key, Text value, OutputCollector output, Reporter reporter) throws IOException String line = value.toString(); StringTokenizer tokenizer = new StringTokenizer(line);/将文章的每个词进行标记 while (tokenizer.hasMoreTokens() word.set(tokenizer.nextToken(); output.collect(word, one);/一个词算一个，以词为分配依据Key，值value为1 ，置入输出 public static class Reduce extends MapReduceBase implements Reducer public void reduce(Text key, Iterator values, OutputCollector output, Reporter reporter) throws IOException int sum = 0; while (values.hasNext() sum += values.next().get();/多分到一个词(key)就加一(迭代器values中的value值) output.collect(key, new IntWritable(sum);/返回词，词的个数 Team Of Programmers / PIG Latin (Hadoop) 熟悉一下PIG的风格（查找最感兴趣页面） PIG实现Map-Reduce（无实际含义） 实际上只负责调度，但实现了一定的松耦合 Visits = load /data/visits as (user, url, time); Visits = foreach Visits generate user, Canonicalize(url), time; Pages = load /data/pages as (url, pagerank); VP = join Visits by url, Pages by url; UserVisits = group VP by user; Sessions = foreach UserVisits generate flatten(FindSessions(*); HappyEndings = filter Sessions by BestIsLast(*); store HappyEndings into /data/happy_endings; a = FOREACH input GENERATE flatten(Map(*); b = GROUP a BY $0; c = FOREACH b GENERATE Reduce(*); Team Of Programmers / Actor Architecture package app.quickstart.hello; import aa.core.Actor; import aa.core.ActorName; import aa.core.CreateActorException; public class Hello extends Actor public Hello() System.out.print(“ Hello“); try ActorName anWorld = create(“app.quickstart.hello.World“); send(anWorld, “world”); catch (CreateActorException e) System.out.println( “ Hello.Hello: “ + e); package app.quickstart.hello; import aa.core.Actor; public class World extends Actor public World() public void world() System.out.println(“ World!“); （调用一个函数world） 内部定位固定、可移动的Actor UAN: Universal Actor Name uan:/9:37 LAN: Location-based Actor Name lan:/47/9:37 这是一种值得学习的整体通信架构。 Team Of Programmers / Thread实现的MapReduce /by package org.ceclipse.cswords.mapreduce; import java.util.Iterator; import java.util.List; public abstract class MapReducer extends Thread protected List output=null; protected Iterator input=null; protected int received=0; public MapReducer(Iterator in,List out) this.input=in; this.output=out; public void run() int i=0;I v=null; while(this.input.hasNext() v=input.next(); Reducer r=this.map(v); r.index=i; r.start(); i+; public abstract Reducer map(I in); public abstract void finished(); public abstract class Reducer extends Thread protected I input=null; protected int index=0; public Reducer(I in) this.input=in; public void run() output.set(this.index, this.reduce(this.input); received+; if(received=output.size()finished(); public abstract O reduce(I in); l对简单运算效率低，但对复杂运算效果明显 l计算40个斐波那契数的效率对比（有波动）： l该算法：3656MS lFor each：5562MS Team Of Programmers / 演员模型的Map-Reduce (Erlang) -module(pmap). -export(pmap/2). pmap(F, L) - S = self(), Pids = lists:map(fun(I) - spawn(fun() - do_fun(S, F, I) end) end, L), gather(Pids). gather(FirstPID|OtherPIDs) - receive FirstPID, Result - Result|gather(OtherPIDs) end; gather() - . do_fun(Parent, F, I) - Parent ! self(), (catch F(I). -module(fib). -export(fib/1). fib(0) - 0; fib(1) - 1; fib(N) when N 1 - fib(N-1) + fib(N-2). Eshell L = lists:seq(0,35). Eshell lists:map(fun(X) - fib:fib(X) end, L). Eshell pmap:pmap(fun(X) - fib:fib(X) end, L). （计算斐波那契数列） 注：左侧代码可以复用 lpmap负责分发进程并建立map存储pid； l每个进程中do_fun向父进程发出自己的进程号 和目标函数计算结果（Reduce）； lgather负责通过递归将每个进程号替换为子进 程发出的函数计算结果。 Team Of Programmers / 实验和观摩总结 技术横向比较结果重点 Hadoop相关技术目前是最成熟的，使用PIG很方便实现大数据处理；但 有框架过重之嫌，代码移植、嵌入现有解决方案都很麻烦；在并非针对 数据的并发中语义不明确； FP演员模型：Erlang表现优秀，处理并行、通信的语法简洁、明确，但 整体语法晦涩；没有第二个类似的虚拟机，其它宿主语言还不成熟； Java原生方案 演员模型 Kilim：调度器没有真正实现多核并行，仅仅实现了消息机制和纤程(fiber)，无助于 多核计算；易于实现代码、逻辑的移植，但并没有打破线程的思想，无助于设计； AA方案：架构清晰，Java风格明显，可以媲美Erlang，效率未知，但还没有其上 的开发模式，演员模型带来的差异性明显，需尝试进一步封装； 线程Map-Reduce方案：效率差，高并发的重量级应用可行。 易行整体方案 超多核虚拟机+单机并行方案（演员模型？） Hadoop全套方案 从海量存储、海量数据处理（搜索引擎？）开始 需封装足够的算法（提供商？） Team Of Programmers / 眼下存在的问题 如何封装并行调度算法 面向对象的语义问题 函数语言的语法问题（以Erlang为例） 去除spawn、receive关键词 去除模式匹配等方面的晦涩语法 领域脚本语言的局限性 PIG LATIN和云数据库上的SQL局限于面向存储的调度 新的设计模式会是怎样的 面向对象的子集消息机制 面向对象的特例面向演员 面向对象的补充 小问题 自然界是否真的存在无穷且无序的集合？ Round Robin如何最优实现“贪心”的目标？ 其他问题？ Team Of Programmers / 在不久的将来所需要的 一种高级语言 支持并行处理，封装并行调度算法 仅支持函数式编程没有代词、变量 Java/C风格 与