1) Import statements and spark Initializtion

Import statements

apache.log4j

import org.apache.log4j._

As spark outputs lots of log data, we can set the log output level to ERROR. So only the error log will be returned. Another commonly used level is 'WARN' which sets log level to return warning errors too.

Logger.getLogger("org").setLevel(Level.ERROR) //prints error logs only

spark.sql

This is the spark 2.0 SQL API. This new API supports a new data structure called DataSets, which is almost all similar to RDD(the one which is always used in spark). And also, this library has various functions from starting the session to ending the session. With this new API, we can even use traditional SQL techniques instead of functional programming techniques.

The SQL library supports only DataSets. If you need to interact the data using RDDs instead of DataSets, you must use it via context Object. We'll be able to use sparkContext object once we create the SparkSession.

  val spark = SparkSession
    .builder()
    .master("local[*]")
    .appName("Books Analyzer Mongo")
    .config("spark.mongodb.input.uri", "mongodb://monster:monster@127.0.0.1:27017/book-miner.bookreviews")
    .config("spark.mongodb.output.uri", "mongodb://monster:monster@127.0.0.1:27017/book-miner.bookreviews")
    .getOrCreate()

SparkSession is the session object which is an abstract class which is used to create the session. builder() is used to build the session. .master("local[*]) defines the master node. For the devices which are opted to be a slave, we can skip this step. local denotes the program is going to run locally. And finally, * denotes the number of clusters. appName() configures the App Name, so once we spark-submit the program, we'll be able to visualize the output on screen with the App Name being displayed. config configures some additional parameters. getOrCreate() creates the session if master, else gets (joins) the master session.

We can stop the sparkSession using stop

spark.stop()

spark.MongoSpark

Spark and Mongo connector.

First, we have to initialize the connection in the spark session object.

    .config("spark.mongodb.input.uri", "mongodb://monster:monster@127.0.0.1:27017/book-miner.bookreviews")
    .config("spark.mongodb.output.uri", "mongodb://monster:monster@127.0.0.1:27017/book-miner.bookreviews")

As said previously to use RDD API, we must use it via sparkContext object. And, to get the whole data from MongoDB as a RDD datastructure, we can pass the sparkContext object to MongoSpark.load function.

  val sc = spark.sparkContext
  //load dataset from mongo db as a rdd
  val rdd = MongoSpark.load(sc).persist()

.persist()

We are persisting the data because we are going to use this data many times. If the data is being used more than once, then persist would be good, also there's an alternative to persist() using cache() but cache() has a data limit.