This recipe shows how Spark supports a wide range of input and output sources. Spark makes it very simple to load and save data in a large number of file formats. Formats range from unstructured, such as text, to semi-structured, such as JSON, to structured, such as SequenceFiles.
To step through this recipe, you will need a running Spark cluster either in pseudo distributed mode or in one of the distributed modes, that is, standalone, YARN, or Mesos. Also, the reader is expected to have an understanding of text files, JSON, CSV, SequenceFiles, and object files.
Load and save a text file as follows:
val input = sc.textFile("hdfs://namenodeHostName:8020/repos/spark/README.md") val wholeInput = sc.wholeTextFiles("file://home/padma/salesFiles") val result = wholeInput.mapValues{value => val nums = value.split (" ").map(x => x.toDouble) nums.sum/nums.size.toDouble} result.saveAsTextFile("/home/Padma/outputFile.txt")For loading a JSON file, the
people.jsoninput file is taken from theSPARK_HOMEfolder whose location is/spark-1.6.0/examples/src/main/resource/people.json. Now, loading and saving a JSON file looks like this:// Loading JSON file import com.fasterxml.jackson.module.scala.DefaultScalaModule import com.fasterxml.jackson.module.scala. experimental.ScalaObjectMapper import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.module.databind. DeserializatiuonFeature ... case class Person(name:String, age:Int) ... val jsonInput = sc.textFile(""hdfs://namenode:9000/data/people.json") val result = jsonInput.flatMap(record => { try{Some(mapper.readValue(record, classOf[Person])) } catch{ case e:Exception => None }} ) result.filter(person => person.age>15).map(mapper.writeValueAsString(_)). saveAsTextFile(output File)To load and save a CSV file, let's take the stocks data:
IBM,20160113,133.5,134.279999,131.100006,131.169998,4672300 GOOG,20160113,730.849976,734.73999,698.609985,700.559998,2468300 MSFT,20160113,53.799999,54.07,51.299999,51.639999,66119000 MSFT,20160112,52.759998,53.099998,52.060001,52.779999,35650700 YHOO,20160113,30.889999,31.17,29.33,29.440001,16593700 . . import java.io.StringReader import au.com.bytecode.opencsv.CSVReader ... case class Stocks(name:String, totalPrice:Long) ... val input = sc.textFile("hdfs://namenodeHostName:8020 /data/stocks.txt") val result = input.map{line => val reader = new CSVReader(new StringReader(line)) reader.readAll().map(x => Stocks(x(0), x(6))) } result.map(stock => Array(stock.name, stock. totalPrice)).mapPartitions {stock => val stringWriter = new StringWriter val csvWriter = new CSVWriter(stringWriter) csvWriter.writeAll(people.toList) Iterator(stringWriter.toString) }.saveAsTextFilehdfs://namenode:9000/CSVOutputFile")Now, let's see the way
sequenceFileis loaded and saved:val data = sc.sequenceFile(inputFile, classOf[Text], classOf[IntWritable]).map{case(x,y) => (x.toString, y.get())} val input = sc.parallelize(List(("Panda",3),("Kay",6), ("Snail",2))) input.saveAsSequenceFilehdfs://namenode:9000/ sequenceOutputFile")
The call to textFile() on the SparkContext with the path to the file loads the text file as RDD. If there exists multiple input parts in the form of a directory then we can use SparkContext.wholeTextFiles(), which returns a pair RDD with the key as the name of the input file. Well, for handling JSON files, the data is loaded as a text file and then it is parsed using a JSON parser. There are a number of JSON libraries available, but in the example we used the Jackson (http://bit.ly/17k6vli) library as it is relatively simple to implement.
Tip
Please refer to other JSON libraries, such as this one: http://bit.ly/1xP8JFK
Loading CSV/TSV data is similar to JSON data, that is, first the data is loaded as text and then processed. Similar to JSON, there are various CSV libraries, but for Scala, we used opencsv (http://opencsv.sourceforge.net). Using CSVReader, the records are parsed and mapped to case class structure. While saving the file, CSVWriter is used to output the file.
When coming to SequenceFile, it is a popular Hadoop format composed of a flat file with key/value pairs. This sequence file implements Hadoop's writable interface. SparkContext.sequenceFile() is the API to load the sequence file in which the parameters classOf[Text] and classOf[IntWritable] indicate the keyClass and valueClass.
As Spark is built on the ecosystem of Hadoop, it can access data through the InputFormat and OutputFormat interfaces used by Hadoop MapReduce, which are available for many common file formats and storage systems (for example, S3, HDFS, Cassandra, HBase, and so on).
Tip
For more information, please refer Hadoop InputFormat (http://hadoop.apache.org/docs/stable/api/org/apache/hadoop/mapred/InputFormat.html) and SequenceFiles (http://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapred/SequenceFileInputFormat.html).
Spark can also interact with any Hadoop supported formats (for both old and new Hadoop file APIs) using newAPIHadoopFile, which takes a path and three classes. The first class represents the input format. The next class is for our key and the final class is the class of our value. The Spark SQL module provides a more efficient API for structured data sources, which includes JSON and Hive.
For more details on Hadoop input and output formats and SequenceFiles input format, please refer to the following:
