-
Book Overview & Buying
-
Table Of Contents
Apache Spark for Data Science Cookbook
By :
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.
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")
people.json input file is taken from the SPARK_HOME folder 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)
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")
sequenceFile is 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.
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).
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:
Change the font size
Change margin width
Change background colour