Apache Spark 2.x Cookbook

        sc.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", "<replace  
          with your key>")

        sc.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey","
          <replace with your secret key>")

        val yelpdata = 
          spark.read.textFile("s3a://sparkcookbook/yelpdata")

        val accessKey = "<your access key>"

        val secretKey = "<your secret key>".replace("/","%2F")

        val bucket = "sparkcookbook"

        val mount = "cookbook"

        dbutils.fs.mount(s"s3a://$accessKey:$secretKey@$bucket",
          s"/mnt/$mount")

        display(dbutils.fs.ls(s"/mnt/$mount"))

Let's look at the key concepts in Databricks Cloud.

What EMR represents is far more than meets the eye. Most of the enterprise workloads are migrating to public clouds at an accelerated pace. Once migrated, these workloads get rearchitected to leverage cloud-based services as opposed to simply using it as Infrastructure as a Service (IaaS). EC2 is an IaaS compute service of AWS, while EMR is the leading Platform as a Service (PaaS) service of AWS, with more big data workloads running on EMR than the alternatives combined. 

Hadoop's core feature is data locality, that is, taking compute to where the data is. AWS disrupts this concept by separating storage and compute. AWS has multiple storage options, including the following:

Amazon S3 is the cheapest and most reliable cloud storage available, and this makes it the first choice, unless there is a compelling reason not to do so. EMR also supports attaching elastic block storage (EBS) volumes to compute instances (EC2) in order to provide a lower latency option.

Which option to choose depends upon what type of cluster is being created. There are two types of clusters:

Let's look at the options shown in step 3:

At the time of writing, Spark's current version is 2.1. Please check the latest version from Spark's download page at http://spark.apache.org/downloads.html. Binaries are developed with the most recent and stable version of Hadoop. To use a specific version of Hadoop, the recommended approach is that you build it from sources, which we will cover in the next recipe.

All the recipes in this book are developed using Ubuntu Linux, but they should work fine on any POSIX environment. Spark expects Java to be installed and the JAVA_HOME environment variable set.

In Linux/Unix systems, there are certain standards for the location of files and directories, which we are going to follow in this book. The following is a quick cheat sheet:

Here are the installation steps:

$ wget http://d3kbcqa49mib13.cloudfront.net/spark-2.1.0-bin-hadoop2.7.tgz

$ tar -zxf spark-2.1.0-bin-hadoop2.7.tgz

$ sudo mv spark-2.1.0-bin-hadoop2.7 spark

$ sudo mv spark/conf/* /etc/spark

$ sudo mkdir -p /opt/infoobjects

$ sudo mv spark /opt/infoobjects/

$ sudo chmod -R 755 /opt/infoobjects/spark

$ cd /opt/infoobjects/spark

$ sudo ln -s /etc/spark conf

$ echo "export PATH=$PATH:/opt/infoobjects/spark/bin" >> /home/hduser/.bashrc

$ sudo mkdir -p /var/log/spark

$ sudo chown -R hduser:hduser /var/log/spark

$ mkdir /tmp/spark

     $ cd /etc/spark
     $ echo "export HADOOP_CONF_DIR=/opt/infoobjects/hadoop/etc/hadoop" >> spark-env.sh
     $ echo "export YARN_CONF_DIR=/opt/infoobjects/hadoop/etc/Hadoop" >> spark-env.sh
     $ echo "export SPARK_LOG_DIR=/var/log/spark" >> spark-env.sh
     $ echo "export SPARK_WORKER_DIR=/tmp/spark" >> spark-env.sh

$ sudo chown -R root:root /opt/infoobjects/spark

The following are the prerequisites for this recipe to work:

The following are the steps to build the Spark source code with Maven:

$ echo "export _JAVA_OPTIONS="-XX:MaxPermSize=1G""  >> 
         /home/hduser/.bashrc

$ wget https://github.com/apache/spark/archive/branch-2.1.zip

$ unzip branch-2.1.zip

        $ mv spark-branch-2.1 spark

$ cd spark

$ mvn -Pyarn -Phadoop-2.7 -Dhadoop.version=2.7.0 -Phive -
        DskipTests clean package

$ sudo mv spark/conf /etc/

$ sudo mv spark /opt/infoobjects/spark

$ sudo chown -R root:root /opt/infoobjects/spark

$ sudo chmod -R 755 /opt/infoobjects/spark

$ cd /opt/infoobjects/spark

$ sudo ln -s /etc/spark conf

$ echo "export PATH=$PATH:/opt/infoobjects/spark/bin" >> 
          /home/hduser/.bashrc

$ sudo mkdir -p /var/log/spark

$ sudo chown -R hduser:hduser /var/log/spark

$ mkdir /tmp/spark

$ cd /etc/spark
$ echo "export HADOOP_CONF_DIR=/opt/infoobjects/hadoop/etc/hadoop" 
       >> spark-env.sh
$ echo "export YARN_CONF_DIR=/opt/infoobjects/hadoop/etc/Hadoop" 
       >> spark-env.sh
$ echo "export SPARK_LOG_DIR=/var/log/spark" >> spark-env.sh
$ echo "export SPARK_WORKER_DIR=/tmp/spark" >> spark-env.sh

The spark-ec2 script comes bundled with Spark and makes it easy to launch, manage, and shut down clusters on Amazon EC2.

Before you start, do the following things: log in to the Amazon AWS account via http://aws.amazon.com.

$ echo "export AWS_ACCESS_KEY_ID="AKIAOD7M2LOWATFXFKQ"" >> /home/hduser/.bashrc
$ echo "export 
       AWS_SECRET_ACCESS_KEY="+Xr4UroVYJxiLiY8DLT4DLT4D4sxc3ijZGMx1D3pfZ2q"" >> 
         /home/hduser/.bashrc
$ echo "export PATH=$PATH:/opt/infoobjects/spark/ec2" >> /home/hduser/.bashrc

$ cd /home/hduser
$ spark-ec2 -k <key-pair> -i <key-file> -s <num-slaves> launch <cluster-name>
<key-pair> - name of EC2 keypair created in AWS
<key-file> the private key file you downloaded
<num-slaves> number of slave nodes to launch
<cluster-name> name of the cluster

$ spark-ec2 -k kp-spark -i /home/hduser/keypairs/kp-spark.pem --hadoop-major-
          version 2  -s 3 launch spark-cluster

$ spark-ec2 -k kp-spark -i /home/hduser/keypairs/kp-spark.pem -z us-east-1b --
          hadoop-major-version 2  -s 3 launch spark-cluster

$ spark-ec2 -k kp-spark -i /home/hduser/keypairs/kp-spark.pem --hadoop-major-
          version 2 -ebs-vol-size 10 -s 3 launch spark-cluster

$ spark-ec2 -k kp-spark -i /home/hduser/keypairs/kp-spark.pem -spot-price=0.15 
          --hadoop-major-version 2  -s 3 launch spark-cluster

$ spark-ec2 -k spark-kp1 -i /home/hduser/kp/spark-kp1.pem  login spark-cluster

$ ephemeral-hdfs/bin/hadoop version
Hadoop 2.0.0-chd4.2.0

$ persistent-hdfs/bin/hadoop version
Hadoop 2.0.0-chd4.2.0

$ cd spark/conf

$ mv log4j.properties.template log4j.properties

$ vi log4j.properties

$ spark-ec2/copydir spark/conf

$ spark-ec2 destroy spark-cluster

Let's consider a cluster of six nodes as an example setup--one master and five slaves (replace them with the actual node names in your cluster):

Master
m1.zettabytes.com
Slaves
s1.zettabytes.com
s2.zettabytes.com
s3.zettabytes.com
s4.zettabytes.com
s5.zettabytes.com

$ echo "export PATH=$PATH:/opt/infoobjects/spark/sbin" >> /home/hduser/.bashrc

hduser@m1.zettabytes.com~] start-master.sh

hduser@s1.zettabytes.com~] spark-class org.apache.spark.deploy.worker.Worker 
          spark://m1.zettabytes.com:7077

hduser@m1.zettabytes.com~] echo "s1.zettabytes.com" >> conf/slaves
hduser@m1.zettabytes.com~] echo "s2.zettabytes.com" >> conf/slaves
hduser@m1.zettabytes.com~] echo "s3.zettabytes.com" >> conf/slaves
hduser@m1.zettabytes.com~] echo "s4.zettabytes.com" >> conf/slaves
hduser@m1.zettabytes.com~] echo "s5.zettabytes.com" >> conf/slaves

Once the slave machine is set up, you can call the following scripts to start/stop the cluster:

        val sparkContext = new SparkContext(new 
          SparkConf().setMaster("spark://m1.zettabytes.com:7077")Setting master URL for 
            spark-shell

$ spark-shell --master spark://master:7077

In standalone mode, Spark follows the master-slave architecture, very much like Hadoop, MapReduce, and YARN. The compute master daemon is called Spark master and runs on one master node. Spark master can be made highly available using ZooKeeper. You can also add more standby masters on the fly if needed.

The compute slave daemon is called a worker, and it exists on each slave node. The worker daemon does the following:

There is, at most, one executor per application, per slave machine.

Both Spark master and the worker are very lightweight. Typically, memory allocation between 500 MB to 1 GB is sufficient. This value can be set in conf/spark-env.sh by setting the SPARK_DAEMON_MEMORY parameter. For example, the following configuration will set the memory to 1 gigabits for both the master and worker daemon. Make sure you have sudo as the super user before running it:

$ echo "export SPARK_DAEMON_MEMORY=1g" >> /opt/infoobjects/spark/conf/spark-env.sh

By default, each slave node has one worker instance running on it. Sometimes, you may have a few machines that are more powerful than others. In that case, you can spawn more than one worker on that machine with the following configuration (only on those machines):

$ echo "export SPARK_WORKER_INSTANCES=2" >> /opt/infoobjects/spark/conf/spark-env.sh

The Spark worker, by default, uses all the cores on the slave machine for its executors. If you would like to limit the number of cores the worker could use, you can set it to the number of your choice (for example, 12), using the following configuration:

$ echo "export SPARK_WORKER_CORES=12" >> /opt/infoobjects/spark/conf/spark-env.sh

The Spark worker, by default, uses all of the available RAM (1 GB for executors). Note that you cannot allocate how much memory each specific executor will use (you can control this from the driver configuration). To assign another value to the total memory (for example, 24 GB) to be used by all the executors combined, execute the following setting:

$ echo "export SPARK_WORKER_MEMORY=24g" >> /opt/infoobjects/spark/conf/spark-env.sh

There are some settings you can do at the driver level:

$ spark-submit --conf spark.cores.max=12

$ spark-submit --conf spark.executor.memory=8g

The following diagram depicts the high-level architecture of a Spark cluster:

To find more configuration options, refer to the following URL:

Mesosphere provides a binary distribution of Mesos. The most recent package of the Mesos distribution can be installed from the Mesosphere repositories by performing the following steps:

$ sudo apt-key adv --keyserver keyserver.ubuntu.com --recv E56151BF 
          DISTRO=$(lsb_release -is | tr '[:upper:]' '[:lower:]') CODENAME=$(lsb_release 
            -cs)
$ sudo vi /etc/apt/sources.list.d/mesosphere.list
deb http://repos.mesosphere.io/Ubuntu trusty main

$ sudo apt-get -y update

$ sudo apt-get -y install mesos

$ hdfs dfs -put spark-2.1.0-bin-hadoop2.7.tgz spark-2.1.0-bin-hadoop2.7.tgz

$ sudo vi spark-env.sh
export MESOS_NATIVE_LIBRARY=/usr/local/lib/libmesos.so
export SPARK_EXECUTOR_URI= hdfs://localhost:9000/user/hduser/spark-2.1.0-bin-
          hadoop2.7.tgz

Val conf = new SparkConf().setMaster("mesos://host:5050")
Val sparkContext = new SparkContext(conf)

$ spark-shell --master mesos://host:5050

Conf.set("spark.mesos.coarse","true")

Running Spark on YARN requires a binary distribution of Spark that has YARN support. In both the Spark installation recipes, we have taken care of this.

HADOOP_CONF_DIR: to write to HDFS
YARN_CONF_DIR: to connect to YARN ResourceManager
$ cd /opt/infoobjects/spark/conf (or /etc/spark)
$ sudo vi spark-env.sh
export HADOOP_CONF_DIR=/opt/infoobjects/hadoop/etc/Hadoop
export YARN_CONF_DIR=/opt/infoobjects/hadoop/etc/hadoop

$ spark-submit --class path.to.your.Class --master yarn --deploy-mode client 
          [options] <app jar> [app options]

Here's an example:

$ spark-submit --class com.infoobjects.TwitterFireHose --master yarn --deploy-
          mode client --num-executors 3 --driver-memory 4g --executor-memory 2g --
            executor-cores 1 target/sparkio.jar 10

$ spark-shell --master yarn --deploy-mode client 

$ spark-submit --class path.to.your.Class --master yarn --deploy-mode cluster 
          [options] <app jar> [app options]

Here's an example:

$ spark-submit --class com.infoobjects.TwitterFireHose --master yarn --deploy-
          mode cluster --num-executors 3 --driver-memory 4g --executor-memory 2g --
            executor-cores 1 target/sparkio.jar 10

Spark applications on YARN run in two modes:

The yarn-cluster mode is recommended for production deployments, while the yarn-client mode is good for development and debugging, where you would like to see the immediate output. There is no need to specify the Spark master in either mode as it's picked from the Hadoop configuration, and the master parameter is either yarn-client or yarn-cluster.

The following figure shows how Spark is run with YARN in the client mode:

The following figure shows how Spark is run with YARN in the cluster mode:

In the YARN mode, the following configuration parameters can be set:

SparkContext is the first object that a Spark program must create to access the cluster. In spark-shell, it is directly accessible via spark.sparkContext. Here's how you can programmatically create SparkContext in your Scala code:

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
val conf = new SparkConf().setAppName("my app").setMaster("master url")
new SparkContext(conf)

SparkContext, though still supported, was more relevant in the case of RDD (covered in the next recipe). As you will see in the rest of the book, different libraries have different wrappers around SparkContext, for example, HiveContext/SQLContext for Spark SQL, StreamingContext for Streaming, and so on. As all the libraries are moving toward DataSet/DataFrame, it makes sense to have a unified entry point for all these libraries as well, and that is SparkSession. SparkSession is available as spark in the spark-shell. Here's how you do it:

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
val sparkSession = SparkSession.builder.master("master url").appName("my app").getOrCreate()

Let's revisit our word count example to understand these five parts. This is how an RDD graph looks for wordCount at the dataset level view:

Basically, this is how the flow goes:

scala> val words = sc.textFile("hdfs://localhost:9000/user/hduser/words")

The following are the five parts of the words RDD:

scala> val wordsFlatMap = words.flatMap(_.split("W+"))

The following are the five parts of the wordsFlatMap RDD:

scala> val wordsMap = wordsFlatMap.map( w => (w,1))

The following are the five parts of the wordsMap RDD:

scala> val wordCount = wordsMap.reduceByKey(_+_)

The following are the five parts of the wordCount RDD:

This is how an RDD graph of wordcount looks at the partition level view: