Apache Spark Download

Regardless of whether you use the development or production version, you must download the latest build from https://spark.apache.org/downloads.html (version 1.6.1 as of this writing).

As shown in Figure 6-1, select Pre-built for Hadoop and later.

Figure 6-1. Apache Spark download page

Spark has a new release every 90 days. For hard-core coders who like to work with the latest builds, try to clone the repository at https://github.com/apache/spark . The instructions for generating the build are available at https://spark.apache.org/docs/latest/building-spark.html . Both the source code and the binary prebuilds are available at this link.

To compile the Spark sources, we need the appropriate versions of Scala and the corresponding SDK. Spark source tar includes the Scala components required.

The Spark development group has done a good job keeping the dependencies. On https://spark.apache.org/docs/latest/building-spark.html , you can see the latest information about it. According to the site, to build Spark with Maven, Java version 6 or higher and Maven 3.0.4 are required.

To uncompress the package, execute the following command:

tar xvf spark-1.6.1-bin-hadoop2.4.tgz

Let’s Kick the Tires

To test the installation, run the following command:

/opt/spark-1.6.1-bin-hadoop2.6/bin/run-example SparkPi 10

You should see an output like the one shown in Figure 6-2, with the line Pi is roughly.

Figure 6-2. Testing Apache Spark

To open a Spark interactive shell, go to the bin directory and run the spark-shell:

$> /bin/spark-shell

You should see output similar to Figure 6-3 (which shows Windows 64-bit so that no one feels left out of this party):

Figure 6-3. The Apache Spark shell

Like all modern shells , the Spark shell includes history. You can access it with the up and down arrows. There are also autocomplete options that you can access by pressing the Tab key.

As you can see, Spark runs in Scala; the Spark shell is a Scala terminal with more features. This chapter’s Scala examples run without problems. You can test, as follows:

scala> val num = 1 to 400000
num: scala.collection.immutable.Range.Inclusive = Range (...
To convert our Range to a RDD (now we see it is that), do the following:
scala> val myRDD = sc.parallelize(num)
myRDD: org.apache.spark.rdd.RDD [Int] = ParallelCollectionRDD [0] at parallelize at <console>

In this case, there is a numeric RDD. Then, as you may guess, you can do all the math operations with Scala data types. Let’s use only the odd numbers:

scala> myRDD.filter (_% 2 != 0) .collect ()
res1: Array [Int] = Array (1, 3, 5, 7, 9 ...)

Spark returns an int array with odd numbers from 1 to 400,000. With this array, you can make all the math operations used with Scala int arrays.

Now, you are inside Spark, where things can be achieved in a big corporate cluster.

Basically, Spark is a framework for processing large volumes of data— in gigabytes, terabytes, or even petabytes. When you work with small data volumes, however, there are many solutions that are more appropriate than Spark.

The two main concepts are the calculations and scale. The effectiveness of the Spark solution lies in making complex calculations over large amounts of data, in an expeditious manner.

Loading a Data File

Upload a text file in Spark within the Spark shell:

scala> val bigfile = sc.textFile ("./big/tooBigFile.txt")

This magically loads the tooBigFile.txt file to Spark, with each line a different entry of the RDD (explained shortly). The RDDs are very versatile in terms of scaling.

If you connect to the Spark master node, you may try to load the file in any of the different machines in the cluster, so you have to ensure that it can be accessed from all worker nodes in the cluster. In general, you always put your files in file systems like HDFS or S3. In local mode, you can add the file directly (e.g., sc.textFile ([path_to_file)). You can use the addFile()SparkContext function to make a file available to all machines in this way:

scala> import org.apache.spark.SparkFiles
scala> val myFile = sc.addFile( "/opt/big/data/path/bigFile.dat" )
scala> val txtFile = sc.textFile (SparkFiles.get("bigFile.txt"))

For example, if you load a (big) input file where each line has a lot of numbers, the first RDD file whose elements are strings (text lines) is not very helpful. To transform the string elements to an array of doubles, use your knowledge of the Scala language:

scala> val myArrays = textFile.map (line => line.split('').map(_. toDouble))

To verify that this is what you wanted, you can use the first() operator on both txtFile and myArrays to see that the first element in the txtFile is a string and in myArrays is an Array[Double].

Loading Data from S3

As part of Amazon support, you have access to a file system called Amazon S3 . To access it, you need the AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY variables (to configure them, see the “Running Spark on EC2” section in this chapter).

For instance, you can use the Amazon examples on a data file from Wikipedia:

scala> val myWiki = sc.textFile ("S3N://bigdatademo/sample/wiki/")

We don’t need to set our AWS credentials as parameters for the Spark shell; this is the general path form for access the S3 file system:

S3N://<AWS ACCESS ID>:<AWS SECRET>@bucket/path

As another example, you can get Wikipedia’s traffic statistics from over the last 16 months at https://aws.amazon.com/datasets/wikipedia-traffic-statistics-v2/ .

SparkContext

Now that you have Spark running on your laptop, let’s start programming in more detail. The driver programs access the Spark core through the SparkContext object, which represents the connection between the cluster and the nodes. In the shell, Spark is always accessed through the sc variable; if you want to learn more about the sc variable, type this:

scala> sc
res0: org.apache.spark.SparkContext = org.apache.spark.SparkContext@4152bd0f

Creating a SparkContext

In a program, you can create a SparkContext object with the following code:

val sparkContext = new SparkContext( masterPath, "applicationName", ["SparkPath (optional)"],["JarsList (optional)"])

It is always possible to hard-code the value of the parameters; however, it is best read from the environment with suitable defaults. This allows you to run the code when you change the host without recompiling it. Using local as the default for the master makes it easy to launch the application in a local testing environment. You must be careful when selecting the defaults. Here’s an example snippet:

import spark.sparkContext._
import scala.util.Properties

val masterPath = Properties.envOrElse("MASTER","local")
val sparkHome = Properties.get("SPARK_HOME")
val jarFiles = Seq(System.get("JARS"))
val sparkContext = new SparkContext(masterPath, "MyAppName", sparkHome, jarFiles)

SparkContext Metadata

The SparkContext object has useful metadata (see Table 6-1); for example, the version number, the application name, and the available memory. If you recall, information about the version is displayed when you start the Spark shell.

Here are some examples of SparkContext metadata to print the Spark version, the application name, the master node’s name and the memory:

$ bin/spark-shell

scala> sc.version
res0: String = 1.6.1

scala> sc.appName
res1: String = Spark shell

scala> sc.master
res2: String = local[*]

scala> sc.getExecutorMemoryStatus
res3: scala.collection.Map[String,(Long, Long)] = Map(localhost:52962 -> (535953408,535953408))

SparkContext Methods

The SparkContext object is the main entry point for your application and your cluster. It is also used for loading and saving data. You can use it to launch additional Spark jobs and remove dependencies. Table 6-2 shows some SparkContext methods, but you can see all the SparkContext attributes and methods at https://spark.apache.org/docs/latest/api/scala/#org.apache.spark.SparkContext $.

Standalone Apps

You can run Spark applications in two ways: from the pretty Spark shell or from a program written in Java, Scala, Python, or R. The difference between the two modes is that when you run standalone applications , you must initialize the SparkContext, as you saw in previous sections.

To run a program in Scala or Java, it is best to use Maven (or Gradle or SBT, whichever you want). You must import the dependency to your project. At the time of this writing, the Spark version is 1.6.1 (version 2 exists, but it’s very new).

The following are the Maven coordinates for version 1.6.1:

groupId = org.apache.spark
artifactId = spark-core_2.10
version = 1.6.1

// All the necessary imports
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
// Create the SparkConf object
val conf = new SparkConf().setMaster("local").setAppName("mySparkApp")

// Create the SparkContext from SparkConf
val sc = new SparkContext(conf)

// We create a RDD with the the-process.txt file contents
val myfile = sc.textFile("the-process.txt")
// Then, we convert the each line text to lowercase
val lowerCase = myFile.map( line => line.toLowerCase)
// We split every line in words (strings separated by spaces)
// As we already know, the split command flattens arrays
val words = lowerCase.flatMap(line => line.split("\s+"))
// Create the tuple (word, frequency), initial frequency is 1
val counts = words.map(word => (word, 1))
// Let’s group the sum of frequencies by word, (easy isn’t?)
val frequency = counts.reduceByKey(_ + _)
// Reverse the tuple to (frequency, word)
val invFrequency = frequency.map(_.swap)
// Take the 20 more frequent and prints it
invFrequency.top(20).foreach(println)

val tinytStopWords = Set("what", "of", "and", "the", "to", "it", "in", "or", "no", "that", "is", "with", "by", "those", "which", "its", "his", "her", "me", "him", "on", "an", "if", "more", "I", "you", "my", "your", "for" )

val words = lowerCase
.flatMap(line => line.split("\s+"))
.filter(! tinyStopWords.contains(_))

// To run it with sbt
sbt clean package
$SPARK_HOME/bin/spark-submit 
--class com.apress.smack.WordFreq 
./target/...(as above) 
./README.md ./wordfreq

// To run it with Maven
mvn clean && mvn compile && mvn package
$SPARK_HOME/bin/spark-submit 
--class com.apress.smack.WordFreq 
./target/WordFreq-0.0.1.jar 
./README.md ./wordfreq

RDD Operations

RDDs have two types of operations: transformations and actions . Transformations are operations that receive one or more RDD as input and return a new RDD. Actions return a result to the driver program and/or store it, and/or trigger a new operation.

If you still get confused and don’t know how to distinguish them, this is the rule: transformations return RDDs; actions don’t.

import org.apache.spark.storage.StorageLevel
val rdd = input.map( foo )
rdd.persist( StorageLevel.DISK_ONLY )
rdd.reduce( bar )
rdd.collect()

Runtime Architecture

Before running Spark on a cluster, it’s important to understand the distributed Spark architecture.

As shown in Figure 6-5, Spark uses a master/slave architecture. The master is called the driver and the slaves are called executors. When running on a single machine, there is a distributed architecture: a driver with several executors. The driver runs in its own Java process, and each executor runs in a separate Java process. This architecture is made on the actor model.

Figure 6-5. Distributed Spark application

The driver and executors set is known as a Spark application. If you have more than one machine, the Spark application must be launched using the cluster manager service. The Spark application architecture is always the same; it does not matter if it’s clustered or not.

In a typical Spark clustered application architecture, each physical machine has its own executor. You will see several strategies to know when an executor dies or goes offline.

Driver

The driver is the process where the SparkContext runs. It is in charge of creating and executing transformations and actions on RDDs. When you run the Spark shell command on your laptop, you are actually creating a driver program. Its first task is to create the SparkContext, called sc. When the driver program dies, the entire application dies.

The following sections explain the two responsibilities in the life of a driver program: dividing a program into tasks and scheduling tasks on executors.

Scheduling Tasks on Executors

Given a physical execution plan, the Spark driver coordinates which tasks are performed by each executor node . When an executor starts operating, it registers itself in the driver, so the driver always has an entire view of all the executor nodes. Each executor is a standalone Java process that can run tasks and store RDDs.

When a program runs, the driver subdivides the program into tasks, sees all the available executor nodes, and tries to balance the workload among them. The driver also knows which part of the data that each node has, in order to rebuild everything at the end.

The driver displays its information on the Web, so that the user can always see what is happening; by default, it runs on port 4040. When you run locally, it can be accessed at http://localhost:4040, as you can see in Figure 6-6 (let’s run the Spark shell and browse it).

Figure 6-6. Spark shell application web UI

Executor

Executors are responsible for running the individual tasks of a given Spark job. Executors are launched when you start the Spark application; they live while the Spark application lives.

Executors have two roles:

• Run the assigned tasks and deliver results to the driver.

• Provide in-memory storage for RDDs. A program called Block Manager runs on each executor and manages memory and RDDs.

When running Spark in local mode, the driver and executors run in the same process. This is for development purposes; it is not recommended in a production environment.

Cluster Manager

Spark depends on a cluster manager to coordinate and launch the executors. The cluster manager that ships with the Spark distribution is called the standalone manager, but it is a pluggable component. You can change it and use a custom cluster manager like Hadoop Yarn, Amazon EC2, or Apache Mesos.

It is important to note that the terms driver and executor are used when talking about the Spark application. When we talk about the cluster manager, we use the terms master and worker. It is important not confuse the terms or to exchange them, because they are different concepts.

Regardless of the cluster manager that you use, Spark provides a single script, called spark-submit, to launch the program. The spark-submit script can connect to different managers through various options and manage the cluster resources that the application needs.

Program Execution

When you run a Spark application on a cluster, these are the steps followed by the program:

1. The user runs the spark-submit shell.

2. The spark-submit shell launches the driver program, and calls the user program’s main() method.

3. The driver program establishes the connection to the cluster manager, which has the slave machines list. Then, the necessary resources are requested to launch the executors.

4. The cluster manager launches executors in each slave node.

5. The driver program analyzes, divides, and distributes the user application, sending each executor its tasks.

6. The tasks run in the executors, calculating and storing the results.

7. The user program ends when the exit() method in the main()method is invoked, or when the SparkContext stop() method is called.

8. The driver program ends the executors and frees the cluster manager’s resources.

Application Deployment

Spark uses the spark-submit tool to send jobs to the cluster manager.

When you run a program in local mode, you only invoke spark-submit passing your script name or jar file as a parameter.

When you run a program in cluster mode, you have to pass additional parameters—for example, the size of each executor process.

The --master flag specifies the cluster URL to which you want to connect. In this case, spark:// means that we are using Spark in stand-alone mode.

For example:

bin/spark-submit --master spark://skynet:7077 --executor-memory 10g Terminator.jar

Here we indicate that we will run our Terminator program in stand-alone cluster mode in the master node called SkyNet and each executor node will have 10 gigabytes of memory.

In addition to the cluster URL, spark-submit has several options to specify how you want to run your application. These options are in of two categories:

• Scheduling data. The amount of resources that each job will have.

• Dependencies. The files and libraries available in the slaves.

Table 6-7 lists and describes some of the spark-submit flags.

Table 6-8 lists a few example values that the --master flag could have.

Now you are able to read this:

$ ./bin/spark-submit 
--master spark://skynet:7077 
--deploy-mode cluster 
--class com.cyberdyne.Terminator 
--name "T1000 model" 
--jars neuralNetwork.jar,geneticAlgorithm.jar 
--total-executor-cores 300 
--executor-memory 10g 
terminator.jar

Here we indicate that we will run our terminator.jar program:

• In stand-alone cluster mode in the SkyNet master node on port 7077

• The driver program is launched in cluster mode

• The main class is com.cyberdyne.Terminator

• The application display name is T1000 model

• The neuralNetwork.jar and geneticAlgorithm.jar files are used

• Each executor node uses 300 cores and has 10 gigabytes of memory

Running in Cluster Mode

This section discusses some of the most common methods to install Spark in cluster mode . On a single laptop, Spark is excellent for developing and testing; but in practice, it is necessary to know how to install Spark with built-in scripts on a dedicated cluster via SSH (Secure Shell). This section covers how to deploy on a cluster in Spark Standalone and with Mesos. This section also covers how to deploy Spark in the cloud with Amazon EC2.

Spark Standalone Mode

When you run bin/start-master.sh, you start an individual master. When you run sbin/start-slaves.sh, you start a worker. The default port of the Spark master is always 8080. Because no one wants to go to each machine and run scripts by hand in each, there is a set of useful scripts in the /bin directory that can help you run your servers.

A prerequisite to use any of the scripts is to have passwordless SSH access from the master to all worker machines. It is always advisable to create a special user to run Spark on all machines. In this book, the examples use the name sparkuser. From the master, you can run the ssh-keygen command to generate the SSH keys. When an RSA key is generated, by default, it is stored in ∼/.ssh/id_rsa.pub. You have to add it to each host in ∼/.ssh/authorized_keys.

The Spark administration scripts require that user names match. If this is not the case, you can configure alternative user names in ∼/.ssh/config.

Now that you have SSH access to the machines, it’s time to configure Spark. There is a template in the conf/spark-env.sh.template directory, which should be copied to conf/spark-env.sh. Table 6-9 lists some of the environment variables.

Once the configuration is made, you must start the cluster. We highly recommend that you install the pssh tool, which is a set of tools that includes pscp SSH. The pscp command makes it easy to secure copying between hosts (although it takes a little while); for example:

pscp -v -r -h conf/slaves -l sparkuser ../spark-1.6.1 ∼/

When you have finished changing the settings, you need to distribute the configuration to workers, as shown here:

pscp -v -r -h conf/slaves -l sparkuser conf/spark-env.sh ∼/spark-1.6.1/conf/spark-env.sh

After you have copied the files, you are ready to start the cluster. Use these scripts: sbin/start-all.sh, sbin/start-master.sh, and sbin/start-slaves.sh.

It is important to note that start-all.sh and start-master.sh are assumed to be running on the master node in the cluster. All scripts start demonizing, so there is no problem running them on the screen:

ssh master bin/start-all.sh

In the event that you get a java.lang.NoClassDefFoundError: scala/ScalaObject error, check that you have Scala installed on the host and that the SCALA_HOME environment variable is set properly.

Spark scripts assume that the master has Spark installed in the same directory as your workers. If this is not the case, you must edit bin/spark-config.sh to point to the correct directories.

Table 6-10 shows the commands provided by Spark to manage the cluster. If you want more information, go to http://spark.apache.org/docs/latest/spark-standalone.html#cluster-launch-scripts .

Now the Spark cluster is running. As shown in Figure 6-7, there is a useful web UI running in the master on port 8080, and in workers running on port 8081. The web UI contains important information about the workers’ current and past jobs.

Figure 6-7. Spark Master UI

Now that you have a cluster up and running, you can do several things with it. As with single host examples, you have the same scripts to run Spark examples on a cluster.

All the example programs are in the examples/src/main/scala/spark/examples/ directory and take the master parameter, which points them to the master IP host. If you are running on the master host, you can run the example:

./run spark.examples.GroupByTest spark://'hostname':7077

If you get a java.lang.UnsupportedClassVersionError error, it is because you need to update the JDK. Always use the supported versions. To check which version compiled your Spark distribution, use the command:

java -verbose -classpath ./core/target/scala-2.11.8/classes/ spark.SparkFiles | head -n 20

Version 49 is JDK 1.5, version 50 is JDK 1.6, and version 60 is JDK 1.7

If you cannot connect to the host, make sure that you have configured your master to listen to all IP addresses.

Running Spark on EC2

The ec2 directory contains the scripts to run a Spark cluster in Amazon EC2. These scripts can be used to run Spark in single or cluster mode. Spark can also run on Elastic MapReduce, which is the Amazon solution for MapReduce cluster management.

The final configuration to run Spark in EC2 is at http://spark.apache.org/docs/latest/ec2-scripts.html .

To begin, you must have EC2 enabled in your Amazon account. It should generate a key pair for the Spark cluster. This can be done at https://portal.aws.amazon.com/gp/aws/securityCredentials . Remember that the key pairs are generated by region. You must make sure to generate them in the same region where you run your hosts. You can also choose to upload a SSH public key instead of generating a new one. They are sensible, so you have to be sure to keep them in a safe place. In our environments, we need two environment variables, AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY, which are available from our EC2 scripts:

export AWS_ACCESS_KEY_ID = {our} AWS access key
export AWS_SECRET_ACCESS_KEY = {our} AWS secret key

There are some scripts provided by Amazon at http://aws.amazon.com/developertools/Amazon-EC2 .

To check that everything is running correctly, type this command:

$ Ec2-describe-regions

That should result in the following output:

REGION  ap-northeast-1  ec2.ap-northeast-1.amazonaws.com
REGION  ap-southeast-1  ec2.ap-southeast-1.amazonaws.com
REGION  ap-southeast-2  ec2.ap-southeast-2.amazonaws.com
REGION  eu-central-1    ec2.eu-central-1.amazonaws.com
REGION  eu-west-1       ec2.eu-west-1.amazonaws.com
REGION  us-east-1       ec2.us-east-1.amazonaws.com
REGION  us-west-1       ec2.us-west-1.amazonaws.com
REGION  us-west-2       ec2.us-west-2.amazonaws.com
REGION  sa-east-1       ec2.sa-east-1.amazonaws.com

The Spark EC2 script automatically generates a group of different security and firewall rules to run the Spark cluster. By default, our Spark cluster is accessible on port 8080. For security, we strongly recommend changing the 0.0.0.0/0 address in the spark_ec2.py script with our public IP address.

To start the cluster, use this command:

./ec2/spark-ec2 -k spark-keypair -i pk-{....}.pem -s 1 launch lePetitCluster

Where {....}.pem indicates the path to our private key.

If you get a “not being reliable to SSH to the master” error, it is because the key can be accessed by others users. Try changing the access permissions to the key so that only one user can read it. If more users can read it, the SSH will refuse it.

If you get the “cannot yet SSH script to the master” error, it is because we are having a race condition; the hosts are reporting them as alive. Try changing the -w in setup_cluster by using a sleep of 120 seconds.

If you get a “transient error while launching a cluster” error, try to run the script with the --resume option.

This makes the scaffolding to make a cluster with a master and a worker node with all default values. The next task is to verify that the firewall rules work. Access the master on port 8080.

The JPS command gives the following information about our cluster:

root @ ip-172-31-45-56 ∼] $ jps
1904 NameNode
2856 Jps
2426 MasterNameNode
2078 SecondaryNodeName

This information is about the name of Spark master, the Hadoop node, and slave nodes.

You can take the example of Pi that you ran on your machine at the beginning of this chapter:

cd spark
bin/run-example SparkPi 10

To terminate the instances, use this command:

ec2/spark-ec2 destroy <cluster name>

To learn all the options that the spark-ec2 command has, run this:

ec2/spark-ec2 -help

There are many types of EC2 instances available; the type has an important impact on the cluster performance. The type can be specified with --instance-type = [typeName]. If you need a lot of memory, you can specify it here.

By default, the same instance type is used for masters and workers. To specify a different type of master you use --master-instance-type = [typeName].

EC2 has also instances of the type GPU, which are very powerful if you run on Envidia graphic cards because they usually have hundreds of GPUs at a very low cost. The GPU instance type is useful when you run a worker, if the master is not used. It is important to note that EC2 performance on the GPU may be lower than when testing locally, because you have more I/O imposed by the hypervisor.

EC2 scripts use the Amazon Machine Images (AMI) provided by the Spark development team; they are enough for most applications.

Running Spark on Mesos

This book devotes an entire chapter on Mesos (Chapter 7), but we must include a special mention in this Spark chapter.

Mesos is a cluster management platform to run multiple distributed applications (or frameworks) on a cluster. Mesos intelligently manages and runs clusters of Spark, Hadoop, Cassandra, Akka, and Kafka. Mesos can run Spark as separate Mesos tasks or run all in a single Mesos task. Mesos quickly scales horizontally to manage clusters beyond the size that individual Python scripts allow.

Mesos has a large number of configuration scripts that you can use. For Ubuntu installations, use configure.ubuntu-lucid-64. In addition to the Spark requirements, you need to ensure that you have the Python C header files installed. Since Mesos must be installed on all of your machines, you can use Mesos to configure other machines:

./configure --prefix=/home/sparkuser/mesos && make && make check && make install            

As with the Spark Standalone mode configuration, you must ensure that Mesos nodes can find each other.

Let’s begin by adding the master hostname to mesossprefix/var/mesos/deploy/masters and all the worker hostnames to mesossprefix/var/mesos/deploy/slaves. Then point the workers to the master in mesossprefix/var/mesos/conf/mesos.conf.

Once Mesos is configured on your machines, you need to configure Spark to run on Mesos. This is as simple as copying the conf/spark-env.sh.template file to conf/spark-env.sh and updating the MESOS_NATIVE_LIBRARY variable to point to the path where Mesos is installed.

Then copy the build to all machines using this secure shell copy command:

pscp -v -r -h  -l sparkuser ./mesos /home/sparkuser/mesos

To start Mesos clusters, use mesosprefix/sbin/mesos-start-cluster.sh. Use mesos://[host]:5050 as the master.

Well, that’s roughly what you have to do to run an Apache Spark cluster on Apache Mesos.

Submitting Our Application

To submit our application to a standalone cluster manager, type this:

spark-submit --master spark://masterNodeHost:masterNodePort appName

Previously, you saw the spark-submit command syntax. Now just change the master node address, and that’s it.

The cluster URL is also displayed in the cluster administrator web UI at http://[masternode]:8080. The host name and port should exactly match the URL present in the web UI. As administrators, we can configure ports other than 7077.

The standalone cluster manager has a --deploy-mode option :

• Client mode (local, default). The driver runs in the same process as spark-submit. If you shut down the spark-submit, your cluster goes down.

• Cluster mode. The driver is launched as a different process on a worker node. You can shut down the machine on which spark-submit is running and everything will continue running.

For each executor node, the --total-executor-cores and --executor-memory parameters specify the number of cores and the available memory. A common error is to ask for more resources than can be assigned. In this case, the cluster manager can’t start the executor nodes. Always check that your cluster has the resources that you state in the parameters.

Configuring Resources

In the standalone cluster manager, the resources are controlled by two variables :

• --executor-memory argument of the spark-submit command

• The total memory on each executor. By default, this value is set to 1GB; unfortunately, in production environments, one gigabyte of memory assigned to each executor won’t be enough.

• --total-executor-cores argument of the spark-submit command

• The total number of cores used on each executor. The default value is unlimited (i.e., the application starts an executor on each available node in the cluster).

To check your current parameters, you can browse the standalone cluster manager’s web UI at http://[masterNode]:8080.

By default, the standalone cluster manager disperses the executors onto the largest number of machines available. For example, suppose you have an eight-node cluster and each machine has four cores. You launch the application with --total-executor-cores 6. Spark will raise only six executors, and tends to use the largest number of machines possible (i.e., an executor on each machine, leaves two nodes without an executor).

If you want to use the fewest number of nodes possible, change the spark.deploy.spreadOut configuration property to false in the conf/spark-defaults.conf configuration file. If we turn off this flag in our example, we will use only two machines: one with four executors and the other with two executors. The other six machines won’t run executors.

These settings affect all the applications on the cluster, so use it with caution.

High Availability

You saw how to specify the cluster mode in the Standalone cluster manager deploy mode, so if the spark-submit process dies, the manager doesn’t also die. This is because the driver runs on one member of the cluster.

You want to keep your manager running while the last node is still standing. To increase manager life, there is a top-notch tool called Apache ZooKeeper, which is discussed in Chapter 7.

Spark Streaming Architecture

Spark Streaming uses the microbatch architecture , in which streaming is considered a continuous series of small batches of data (see Figure 6-8). The magic of Spark Streaming is receiving a continuous flow and splitting it into small data chunks.

Figure 6-8. A DStream is an RDD sequence

Batches are generated at regular time intervals; if two data chunks come in the same time window, they are included in the same data batch. The batch size is determined by the parameter batch interval, which usually has a value between 500 milliseconds and several seconds. The developer specifies this value.

As you can see in Figure 6-9, the primary Spark Streaming task is to receive a data stream from multiple sources, and then build a DStream, which is a sequence of RDDs. Each RDD corresponds to a time slice of the original flow.

Figure 6-9. Spark Streaming operation

We can create DStreams from input sources or through applying transformations to other DStreams. The DStreams support most of the operations that RDDs support. Additionally, DStreams have “stateful” operations to aggregate data across time.

In addition to transformations, DStreams support output operations, which are similar to actions in the aspect that RDDs write to external systems; but Spark Streaming batches run periodically, producing the output batch.

For each input source, Spark launches streaming receivers, which are tasks that run on executors, gather data from information sources, and store it in RDDs. The receivers are also responsible for replicating data among other executors to support fault tolerance. This data is stored in the memory of executors in the same way as the RDDs. As shown in Figure 6-10, the StreamingContext in the driver program periodically runs Spark jobs to process this data and combine them with the new RDDs.

Figure 6-10. Spark Streaming execution with Spark components

The same fault tolerance properties offered by RDDs are also offered with DStreams. In the event of a failure, Spark can recalculate the DStreams at any point in time. However, the recalculation can take time, especially if the rebuild goes back to the beginning of the execution.

Spark Streaming provides a mechanism called checkpointingthat periodically saves the state to a reliable file system. Typically, a checkpoint occurs every five to ten data batches. When a failure occurs, Spark restores from the last checkpoint.

Transformations

Transformations on DStreams can be grouped into stateless or stateful.

• Stateless. The data of each processing batch doesn’t depend on the data from previous batches. These include the RDD transformations such as map(), reduce(), and filter().

• Stateful. The data or intermediate results from previous batches are used to calculate the current batch’s results. These include transformations based on sliding windows and tracking status over time.

window(windowLength, slideInterval)

val wind = lines.window(Seconds(30),Seconds(10));
wind.foreachRDD( rdd => { rdd.foreach( x => println(x+ " ")) })

10 10 20 20 10 30 20 30 40 // drops 10

countByWindow( windowLength, slideInterval )

lines.countByWindow( Seconds(30), Seconds(10) ).print()

1 2 3 3

24/7 Spark Streaming

Spark Streaming provides a mechanism to ensure fault tolerance. If the input data is stored reliably, Spark Streaming always calculates the result from it, providing the correct semantics (i.e., as if the data had been processed without failing nodes).

Spark Streaming applications that run 24/7 need special installation. The first step is to enable checkpointing on a reliable storage system: HDFS or Amazon S3. Also, note that you must deal with the driver program fault tolerance, changing some portion of the code.

Checkpointing

Checkpointing is the primary mechanism to enable fault tolerance in Spark Streaming. Spark Streaming allows you to periodically save the application data in a reliable file system, such as HDFS, or Amazon S3. Checkpointing has two purposes:

• Limits the state to be recomputed when a fault occurs. Spark Streaming recalculates the state using a lineage graph of transformations; checkpointing tells you how far back you should go.

• Driver program fault tolerance. If the driver program of a streaming application crashes, you can start it again and recover from the last checkpoint. Spark Streaming reads how much had processed and will resume from that point.

For these reasons checkpointing is important when you run production Spark Streaming applications.

Note that when running in local mode, Spark Streaming won’t run a stateful operation if you don’t have checkpointing enabled. In this case, you can use a local file system. In a production environment, you should use a replicated file system, such as HDFS, Amazon S3, or NFS.

Spark Streaming Performance

Spark Streaming has specific considerations in addition to Spark performance considerations.