Learning Spark by Holden Karau O'Reilly

好书，好书。。。

1. /**
   
2. * Illustrates lack of caching
   
3. */
   
4. package com.oreilly.learningsparkexamples.scala
   
5. 
   
6. import org.apache.spark._
   
7. 
   
8. object BasicMapNoCache {
   
9.     def main(args: Array[String]) {
   
10.       val master = args.length match {
    
11.         case x: Int if x > 0 => args(0)
    
12.         case _ => "local"
    
13.       }
    
14.       val sc = new SparkContext(master, "BasicMapNoCache", System.getenv("SPARK_HOME"))
    
15.       val input = sc.parallelize(List(1,2,3,4))
    
16.       val result = input.map(x => x*x)
    
17.       // will compute result twice
    
18.       println(result.count())
    
19.       println(result.collect().mkString(","))
    
20.     }
    
21. }

复制代码

1. /**
   
2. * Illustrates a simple map partition in Scala
   
3. */
   
4. package com.oreilly.learningsparkexamples.scala
   
5. 
   
6. import org.apache.spark._
   
7. 
   
8. import org.eclipse.jetty.client.ContentExchange
   
9. import org.eclipse.jetty.client.HttpClient
   
10. 
    
11. object BasicMapPartitions {
    
12.     def main(args: Array[String]) {
    
13.       val master = args.length match {
    
14.         case x: Int if x > 0 => args(0)
    
15.         case _ => "local"
    
16.       }
    
17.       val sc = new SparkContext(master, "BasicMapPartitions", System.getenv("SPARK_HOME"))
    
18.       val input = sc.parallelize(List("KK6JKQ", "Ve3UoW", "kk6jlk", "W6BB"))
    
19.       val result = input.mapPartitions{
    
20.         signs =>
    
21.         val client = new HttpClient()
    
22.         client.start()
    
23.         signs.map {sign =>
    
24.           val exchange = new ContentExchange(true);
    
25.           exchange.setURL(s"http://qrzcq.com/call/${sign}")
    
26.           client.send(exchange)
    
27.           exchange
    
28.         }.map{ exchange =>
    
29.           exchange.waitForDone();
    
30.           exchange.getResponseContent()
    
31.         }
    
32.       }
    
33.       println(result.collect().mkString(","))
    
34.     }
    
35. }

复制代码

1. /**
   
2. * Illustrates a simple map the filter in Scala
   
3. */
   
4. package com.oreilly.learningsparkexamples.scala
   
5. 
   
6. import org.apache.spark._
   
7. 
   
8. object BasicMapThenFilter {
   
9.     def main(args: Array[String]) {
   
10.       val master = args.length match {
    
11.         case x: Int if x > 0 => args(0)
    
12.         case _ => "local"
    
13.       }
    
14.       val sc = new SparkContext(master, "BasicMap", System.getenv("SPARK_HOME"))
    
15.       val input = sc.parallelize(List(1,2,3,4))
    
16.       val squared = input.map(x => x*x)
    
17.       val result = squared.filter(x => x != 1)
    
18.       println(result.collect().mkString(","))
    
19.     }
    
20. }

复制代码

1. /**
   
2. * Illustrates a simple map partition to parse CSV data in Scala
   
3. */
   
4. package com.oreilly.learningsparkexamples.scala
   
5. 
   
6. import java.io.StringReader
   
7. import java.io.StringWriter
   
8. 
   
9. import org.apache.spark._
   
10. import play.api.libs.json._
    
11. import play.api.libs.functional.syntax._
    
12. import scala.util.parsing.json.JSON
    
13. import scala.collection.JavaConversions._
    
14. 
    
15. import au.com.bytecode.opencsv.CSVReader
    
16. import au.com.bytecode.opencsv.CSVWriter
    
17. 
    
18. object BasicParseCsv {
    
19.   case class Person(name: String, favouriteAnimal: String)
    
20. 
    
21.   def main(args: Array[String]) {
    
22.     if (args.length < 3) {
    
23.       println("Usage: [sparkmaster] [inputfile] [outputfile]")
    
24.       exit(1)
    
25.     }
    
26.     val master = args(0)
    
27.     val inputFile = args(1)
    
28.     val outputFile = args(2)
    
29.     val sc = new SparkContext(master, "BasicParseCsv", System.getenv("SPARK_HOME"))
    
30.     val input = sc.textFile(inputFile)
    
31.     val result = input.map{ line =>
    
32.       val reader = new CSVReader(new StringReader(line));
    
33.       reader.readNext();
    
34.     }
    
35.     val people = result.map(x => Person(x(0), x(1)))
    
36.     val pandaLovers = people.filter(person => person.favouriteAnimal == "panda")
    
37.     pandaLovers.map(person => List(person.name, person.favouriteAnimal).toArray).mapPartitions{people =>
    
38.       val stringWriter = new StringWriter();
    
39.       val csvWriter = new CSVWriter(stringWriter);
    
40.       csvWriter.writeAll(people.toList)
    
41.       Iterator(stringWriter.toString)
    
42.     }.saveAsTextFile(outputFile)
    
43.   }
    
44. }

复制代码

1. /**
   
2. * Illustrates a simple map the filter in Scala
   
3. */
   
4. package com.oreilly.learningsparkexamples.scala
   
5. 
   
6. import org.apache.spark._
   
7. 
   
8. object BasicMapThenFilter {
   
9.     def main(args: Array[String]) {
   
10.       val master = args.length match {
    
11.         case x: Int if x > 0 => args(0)
    
12.         case _ => "local"
    
13.       }
    
14.       val sc = new SparkContext(master, "BasicMap", System.getenv("SPARK_HOME"))
    
15.       val input = sc.parallelize(List(1,2,3,4))
    
16.       val squared = input.map(x => x*x)
    
17.       val result = squared.filter(x => x != 1)
    
18.       println(result.collect().mkString(","))
    
19.     }
    
20. }

复制代码