Learning Spark by Holden Karau O'Reilly

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 partition to parse JSON data in Scala
   
3. * Loads the data into a case class with the name and a boolean flag
   
4. * if the person loves pandas.
   
5. */
   
6. package com.oreilly.learningsparkexamples.scala
   
7. 
   
8. import org.apache.spark._
   
9. import play.api.libs.json._
   
10. import play.api.libs.functional.syntax._
    
11. 
    
12. object BasicParseJson {
    
13.   case class Person(name: String, lovesPandas: Boolean)
    
14.   implicit val personReads = Json.format[Person]
    
15. 
    
16.   def main(args: Array[String]) {
    
17.     if (args.length < 3) {
    
18.       println("Usage: [sparkmaster] [inputfile] [outputfile]")
    
19.       exit(1)
    
20.       }
    
21.     val master = args(0)
    
22.     val inputFile = args(1)
    
23.     val outputFile = args(2)
    
24.     val sc = new SparkContext(master, "BasicParseJson", System.getenv("SPARK_HOME"))
    
25.     val input = sc.textFile(inputFile)
    
26.     val parsed = input.map(Json.parse(_))
    
27.     // We use asOpt combined with flatMap so that if it fails to parse we
    
28.     // get back a None and the flatMap essentially skips the result.
    
29.     val result = parsed.flatMap(record => personReads.reads(record).asOpt)
    
30.     result.filter(_.lovesPandas).map(Json.toJson(_)).saveAsTextFile(outputFile)
    
31.     }
    
32. }

复制代码

1. /**
   
2. * Illustrates a simple map partition to parse JSON data in Scala
   
3. * Loads the data into a case class with the name and a boolean flag
   
4. * if the person loves pandas.
   
5. */
   
6. package com.oreilly.learningsparkexamples.scala
   
7. 
   
8. import org.apache.spark._
   
9. import com.fasterxml.jackson.module.scala.DefaultScalaModule
   
10. import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
    
11. import com.fasterxml.jackson.databind.ObjectMapper
    
12. import com.fasterxml.jackson.databind.DeserializationFeature
    
13. 
    
14. 
    
15. 
    
16. case class Person(name: String, lovesPandas: Boolean) // Note: must be a top level class
    
17. 
    
18. object BasicParseJsonWithJackson {
    
19. 
    
20.   def main(args: Array[String]) {
    
21.     if (args.length < 3) {
    
22.       println("Usage: [sparkmaster] [inputfile] [outputfile]")
    
23.       exit(1)
    
24.       }
    
25.     val master = args(0)
    
26.     val inputFile = args(1)
    
27.     val outputFile = args(2)
    
28.     val sc = new SparkContext(master, "BasicParseJsonWithJackson", System.getenv("SPARK_HOME"))
    
29.     val input = sc.textFile(inputFile)
    
30. 
    
31.     // Parse it into a specific case class. We use mapPartitions beacuse:
    
32.     // (a) ObjectMapper is not serializable so we either create a singleton object encapsulating ObjectMapper
    
33.     //     on the driver and have to send data back to the driver to go through the singleton object.
    
34.     //     Alternatively we can let each node create its own ObjectMapper but that's expensive in a map
    
35.     // (b) To solve for creating an ObjectMapper on each node without being too expensive we create one per
    
36.     //     partition with mapPartitions. Solves serialization and object creation performance hit.
    
37.     val result = input.mapPartitions(records => {
    
38.         // mapper object created on each executor node
    
39.         val mapper = new ObjectMapper with ScalaObjectMapper
    
40.         mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
    
41.         mapper.registerModule(DefaultScalaModule)
    
42.         // We use flatMap to handle errors
    
43.         // by returning an empty list (None) if we encounter an issue and a
    
44.         // list with one element if everything is ok (Some(_)).
    
45.         records.flatMap(record => {
    
46.           try {
    
47.             Some(mapper.readValue(record, classOf[Person]))
    
48.           } catch {
    
49.             case e: Exception => None
    
50.           }
    
51.         })
    
52.     }, true)
    
53.     result.filter(_.lovesPandas).mapPartitions(records => {
    
54.       val mapper = new ObjectMapper with ScalaObjectMapper
    
55.       mapper.registerModule(DefaultScalaModule)
    
56.       records.map(mapper.writeValueAsString(_))
    
57.     })
    
58.       .saveAsTextFile(outputFile)
    
59.     }
    
60. }

复制代码

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. 
   
8. import org.apache.spark._
   
9. import play.api.libs.json._
   
10. import play.api.libs.functional.syntax._
    
11. import scala.util.parsing.json.JSON
    
12. import scala.collection.JavaConversions._
    
13. import au.com.bytecode.opencsv.CSVReader
    
14. 
    
15. object BasicParseWholeFileCsv {
    
16.   def main(args: Array[String]) {
    
17.     if (args.length < 2) {
    
18.       println("Usage: [sparkmaster] [inputfile]")
    
19.       exit(1)
    
20.     }
    
21.     val master = args(0)
    
22.     val inputFile = args(1)
    
23.     val sc = new SparkContext(master, "BasicParseWholeFileCsv", System.getenv("SPARK_HOME"))
    
24.     val input = sc.wholeTextFiles(inputFile)
    
25.     val result = input.flatMap{ case (_, txt) =>
    
26.       val reader = new CSVReader(new StringReader(txt));
    
27.       reader.readAll()
    
28.     }
    
29.     println(result.collect().map(_.toList).mkString(","))
    
30.     }
    
31. }

复制代码

感谢楼主的分享~~~~

谢谢分享

kankanakanka

Thanks for sharing