常见计算框架算子层对比

背景

      前段时间在为内部自研的计算框架设计算子层，参考对比了一些开源的计算框架的算子层，本文做一个粗粒度的梳理。

      下面这张图是我对计算框架抽象层次的一个拆分，具体可以参考上周日杭州Spark meetup上我做的Spark SQL分享 slides。

Pig-latin

[java] view plaincopy

1. A = load 'xx' AS (c1:int, c2:chararray, c3:float) 
2. B = GROUP A BY c1 
3. C = FOREACH B GENERATE group, COUNT(A) 
4. C = FOREACH B GENERATE $0. $1.c2 
5.  
6. X = COGROUP A by a1, B BY b1 
7. Y = JOIN A by a1 (LEFT|FULL|LEFT OUTER), B BY b1 

Cascading

[java] view plaincopy

1. // define source and sink Taps. 
2. Scheme sourceScheme = new TextLine( new Fields( "line" ) ); 
3. Tap source = new Hfs( sourceScheme, inputPath ); 
4. Scheme sinkScheme = new TextLine( new Fields( "word", "count" ) ); 
5. Tap sink = new Hfs( sinkScheme, outputPath, SinkMode.REPLACE ); 
6.  
7. // the 'head' of the pipe assembly 
8. Pipe assembly = new Pipe( "wordcount" ); 
9.  
10. // For each input Tuple 
11. // parse out each word into a new Tuple with the field name "word" 
12. // regular expressions are optional in Cascading 
13. String regex = "(?<!\\pL)(?=\\pL)[^ ]*(?<=\\pL)(?!\\pL)"; 
14. Function function = new RegexGenerator( new Fields( "word" ), regex ); 
15. assembly = new Each( assembly, new Fields( "line" ), function ); 
16.  
17. // group the Tuple stream by the "word" value 
18. assembly = new GroupBy( assembly, new Fields( "word" ) ); 
19.  
20. // For every Tuple group 
21. // count the number of occurrences of "word" and store result in 
22. // a field named "count" 
23. Aggregator count = new Count( new Fields( "count" ) ); 
24. assembly = new Every( assembly, count ); 
25.  
26. // initialize app properties, tell Hadoop which jar file to use 
27. Properties properties = new Properties(); 
28. AppProps.setApplicationJarClass( properties, Main.class ); 
29.  
30. // plan a new Flow from the assembly using the source and sink Taps 
31. // with the above properties 
32. FlowConnector flowConnector = new HadoopFlowConnector( properties ); 
33. Flow flow = flowConnector.connect( "word-count", source, sink, assembly ); 
34.  
35. // execute the flow, block until complete 
36. flow.complete();<strong> 
37. </strong> 

Trident

[java] view plaincopy

1. TridentState urlToTweeters = 
2.        topology.newStaticState(getUrlToTweetersState()); 
3. TridentState tweetersToFollowers = 
4.        topology.newStaticState(getTweeterToFollowersState()); 
5.  
6. topology.newDRPCStream("reach") 
7.        .stateQuery(urlToTweeters, new Fields("args"), new MapGet(), new Fields("tweeters")) 
8.        .each(new Fields("tweeters"), new ExpandList(), new Fields("tweeter")) 
9.        .shuffle() 
10.        .stateQuery(tweetersToFollowers, new Fields("tweeter"), new MapGet(), new Fields("followers")) 
11.        .parallelismHint(200) 
12.        .each(new Fields("followers"), new ExpandList(), new Fields("follower")) 
13.        .groupBy(new Fields("follower")) 
14.        .aggregate(new One(), new Fields("one")) 
15.        .parallelismHint(20) 
16.        .aggregate(new Count(), new Fields("reach")); 

RDD

[java] view plaincopy

1. scala> val textFile = sc.textFile("README.md") 
2. textFile: spark.RDD[String] = spark.MappedRDD@2ee9b6e3 
3.  
4. scala> textFile.count() // Number of items in this RDD 
5. res0: Long = 126 
6.  
7. scala> textFile.first() // First item in this RDD 
8. res1: String = # Apache Spark 
9.  
10. scala> val linesWithSpark = textFile.filter(line => line.contains("Spark")) 
11. linesWithSpark: spark.RDD[String] = spark.FilteredRDD@7dd4af09 
12.  
13. scala> textFile.filter(line => line.contains("Spark")).count() // How many lines contain "Spark"? 
14. res3: Long = 15 
15.  
16. scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b) 
17. res4: Long = 15 
18.  
19. scala> val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b) 
20. wordCounts: spark.RDD[(String, Int)] = spark.ShuffledAggregatedRDD@71f027b8 
21.  
22. scala> wordCounts.collect() 
23. res6: Array[(String, Int)] = Array((means,1), (under,2), (this,3), (Because,1), (Python,2), (agree,1), (cluster.,1), ...)

SchemaRDD

[java] view plaincopy

1. val sqlContext = new org.apache.spark.sql.SQLContext(sc) 
2. // createSchemaRDD is used to implicitly convert an RDD to a SchemaRDD. 
3. import sqlContext.createSchemaRDD 
4.  
5. // Define the schema using a case class. 
6. case class Person(name: String, age: Int) 
7.  
8. // Create an RDD of Person objects and register it as a table. 
9. val people = sc.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)) 
10. people.registerAsTable("people") 
11.  
12. // SQL statements can be run by using the sql methods provided by sqlContext. 
13. val teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19") 
14.  
15. // DSL: where(), select(), as(), join(), limit(), groupBy(), orderBy() etc. 
16. val teenagers = people.where('age >= 10).where('age <= 19).select('name) 
17. teenagers.map(t => "Name: " + t(0)).collect().foreach(println)

Apache Crunch

[java] view plaincopy

1. public class WordCount extends Configured implements Tool, Serializable { 
2.   public int run(String[] args) throws Exception { 
3.     // Create an object to coordinate pipeline creation and execution. 
4.     Pipeline pipeline = new MRPipeline(WordCount.class, getConf()); 
5.     // Reference a given text file as a collection of Strings. 
6.     PCollection<String> lines = pipeline.readTextFile(args[0]); 
7.  
8.     PCollection<String> words = lines.parallelDo(new DoFn<String, String>() { 
9.       public void process(String line, Emitter<String> emitter) { 
10.         for (String word : line.split("\\s+")) { 
11.           emitter.emit(word); 
12.         } 
13.       } 
14.     }, Writables.strings()); // Indicates the serialization format 
15.  
16.     PTable<String, Long> counts = words.count(); 
17.     // Instruct the pipeline to write the resulting counts to a text file. 
18.     pipeline.writeTextFile(counts, args[1]); 
19.     // Execute the pipeline as a MapReduce. 
20.     PipelineResult result = pipeline.done(); 
21.  
22.     return result.succeeded() ? 0 : 1; 
23.   } 
24.  
25.   public static void main(String[] args) throws Exception { 
26.     int result = ToolRunner.run(new Configuration(), new WordCount(), args); 
27.     System.exit(result); 
28.   } 
29. } 

总结

      最后这张图展示了Hadoop之上各种Data Pipeline项目的实现层次对比： 

  原文链接：http://blog.csdn.net/pelick/article/details/39076223