DataProc - Spark Cluster on GCP in minutes

I’ve decided to try out running Apache Spark on various ways on Google Cloud Platform,

I’ll tell you a bit about my experience and the ways to perform all of the needed actions each way.

For the first way, I’ll start with the easiest way, using Google’s DataProc service (currently on Beta). If some of you are using Amazon’s AWS it’s the equivalent of their EMR (Elastic MapReduce) service, you can launch a Spark cluster with a GUI tool in the Google cloud console, REST API or via command line tool (I’ll show all of the possibilities next).

First you’ll need to create a Google Cloud account, you can do so in the next link, you get a free trial of 300$ of credits, it will be more than enough for all of your tests. Once you've created the account we can start using the web console to launch the cluster.

You might want to prepare in advance two things (If not, you’ll get the default which are fine too):

1. Creating a “Cloud Storage staging bucket” to stage files, such as Hadoop jars, between client machines and the cluster. If not specified, a default bucket is used.
2. Creating a Network for the Spark cluster the Compute Engine network to use for the cluster. If not specified, the default network will be chosen for you.
   I’ve added a screenshot of the network I’ve created that’s called “spark-cluster-network”, and opened up only the relevant Firewall rules (both for connecting to the cluster and to being able to see the UI features of the Spark cluster).

The next step will be to launch the cluster with DataProc, there are 3 ways to do that:

1. A GUI tool of DataProc on your Cloud console: To get to the DataProc menu we’ll need to follow the next steps:

On the main console menu find the DataProc service:

Then you can create a new cluster, with all of the parameters we’ve talked about before, in our case it’s called “cluster-1”.

After giving the launch command the cluster was up and running after ~45 seconds and I was able to connect to it via SSH:

And you can see that Apache Spark was already pre-installed at “/var/lib/spark”.

Just to check that all is running well I ran spark-shell:

But note a very important thing, you’ll need to launch each application (spark-shell included) with the config parameter to override the Dynamic Allocation feature of YARN.

I got this problem when I launched the job without this configuration, that all of a sudden i would lose executors during the spark-shell was running: (you can see it in the spark master UI that the executors were removed)

Thanks to Vadim Solovey, Dynamic allocation causes Spark to relinquish idle executors back to YARN and unfortunately at the moment spark prints that spammy but harmless "lost executor" message. This was the classical problem of spark on YARN where spark originally paralyzed clusters it ran on because it would grab the maximum number of containers it thought it needed and then never give them up.

With dynamic allocation, when you start a long job, spark quickly allocates new containers (with something like exponential ramp-up to quickly be able to fill a full YARN cluster within a couple minutes) and when idle, relinquishes executors with the same ramp-down at an interval of about 60 seconds (if idle for 60 seconds, relinquish some executors).

If you want to disable dynamic allocation you can run: “spark-shell --conf spark.dynamicAllocation.enabled=false”

or,

“gcloud beta dataproc jobs submit spark --properties spark.dynamicAllocation.enabled=false --cluster <your-cluster> application.jar”

Alternatively, if you specify a fixed number of executors, it should also automatically disable dynamic allocation:
“spark-shell --conf spark.executor.instances=123”

or,

“gcloud beta dataproc jobs submit spark --properties spark.executor.instances=123 --cluster <your-cluster> application.jar”

Some other useful configuration you probably would like to run with are:

“--conf spark.logConf=true --conf spark.logConf=true --conf spark.ui.killEnabled=true”

2. Command line tool that you’ll need to install the cloud SDK for that on your management machine.
   For example: (You can also generate the command from the GUI tool)

gcloud beta dataproc clusters create cluster-1 --zone us-central1-a --master-machine-type n1-standard-4 --master-boot-disk-size 500 --num-workers 2 --worker-machine-type n1-standard-4 --worker-boot-disk-size 500 --num-preemptible-workers 2 --image-version 0.2 --project gcp-tools-01

3. REST API that you can launch a cluster as well from.

POST /v1beta1/projects/gcp-tools-01/clusters/
{
 "clusterName": "cluster-1",
 "projectId": "gcp-tools-01",
 "configuration": {
   "configurationBucket": "",
   "gceClusterConfiguration": {
     "networkUri": "https://www.googleapis.com/compute/v1/projects/gcp-tools-01/global/networks/default",
     "zoneUri": "https://www.googleapis.com/compute/v1/projects/gcp-tools-01/zones/us-central1-a"
   },
   "masterConfiguration": {
     "numInstances": 1,
     "machineTypeUri": "https://www.googleapis.com/compute/v1/projects/gcp-tools-01/zones/us-central1-a/machineTypes/n1-standard-4",
     "diskConfiguration": {
       "bootDiskSizeGb": 500,
       "numLocalSsds": 0
     }
   },
   "workerConfiguration": {
     "numInstances": 2,
     "machineTypeUri": "https://www.googleapis.com/compute/v1/projects/gcp-tools-01/zones/us-central1-a/machineTypes/n1-standard-4",
     "diskConfiguration": {
       "bootDiskSizeGb": 500,
       "numLocalSsds": 0
     }
   },
   "secondaryWorkerConfiguration": {
     "numInstances": "2",
     "isPreemptible": true
   },
   "softwareConfiguration": {
     "imageVersion": "0.2"
   }
 }
}

You can also create an initialization script, A list of scripts to be executed during initialization of the cluster. Each must be a GCS file with a gs:// prefix.

Here’s a list of all the DataProc initialization actions by Google at their GitHub account: https://github.com/GoogleCloudPlatform/dataproc-initialization-actions

And here are some more API docs of the way to create your own init-actions https://cloud.google.com/dataproc/init-actions.

Specific network configuration adjustments to be made:

Because we’ve created a network of our own that is now open for the outer world, we’ll need to open some vital port to expose the Web UI of some Spark services, Allowing the TCP ports of: 4040, 18080, 8088, 19888 will allow you the next services.
(Note: You might need to open other ports for the outer world if you choose to run other frameworks than the ones listed below:

Spark Master UI: http://<Master IP Address>:4040

Spark History Server: http://<Master IP Address>:18080

Yarn Application Master: http://<Master IP Address>:8088/cluster

Hadoop Job History Server: http://<Master IP Address>:19888/jobhistory

Conclusion:

There you have it, in a manner of minutes, even without knowing anything about DataProc / Spark cluster launching you’ll have a running environment on Google Cloud Platform.

In the process you can also choose a certain amount of Preemptible VMs as more executors that will be cheaper than Compute Engine VMs, they will be launched as part of your cluster.

With all of that said, It’s a paid service, and you need to take that in consideration, I think that most of the time you would want to run DataProc is for a pre-defined period of time jobs, that you’ll need to launch a cluster for, do a computation load and then destroy the cluster, and not for a Forever running Spark cluster that you might want to make adjustments and Install additional tools on.

So what are the other options to running Apache Spark on GCP?
next we will show bdutil by Google, A command line tool that provided API to manage Hadoop and Spark tool on GCP and another way to Launch a Mesos cluster on top of GCP and then running Apache Spark on it,

But that will be in future blog posts.

If you have any further questions,
please leave comments, hope this helps you get into the Spark world on GCP.

Hacking and bending Spark Standalone's execution model

Hi all,

After switching from working with a YARN Spark cluster to a Standalone spark cluster, we've encountered some resource management problems,
Our main goal was to run multiple jobs on the cluster that will share the resources and will be in maximum resource utilization.

A bit about the environment, we were running our Spark cluster on Amazon's EMR, and we switched to running the standalone cluster on simple EC2 servers.

The logical thing that we knew was the next architectural explanation:
- There is a Master server: r3.xlarge (ec2 instance type) that controls the cluster.
- Core - the server on with the Spark "Driver" program would run on.
- Multiple Slaves (which are called also "Executors") r3.xlarge too. 

It changed a bit in the standalone cluster, only the "Core" component was dropped, all of the other things stayed the same.
Each Executor had a certain amount of physical cores (CPUs), 4 in the case of r3.xlarge and we would request in "spark-submit" the amount of executors we want, and got the product of Executors * CPUs. (via the parmeters: "--num-executors" and "--executor-cores")

All was great... 
If we took a deep dive to the structure of a single slave machine we would see:

The spark "Worker" is the whole machine, it has 4 cores (CPU's), and a total memory of 31GB (in the case of r3.xlarge).
Again sounds great by now....The topology of the Spark Standalone cluster looks like this now:

One master, many slave machines (workers), and each worker has 4 cores. Each worker also has it's RAM memory, in our case like we said 31GB.

Here comes the problem when we needed to execute multiple Spark applications on the same cluster and share the resources (note that different applications may require different amount of cores and memory).

Let's talk about an example with real numbers, We have a cluster of 10 machines (workers), each has 4 cores, so in total we have 40 cores.
In "spark-submit" in standalone mode, you can pass "--total-executor-cores" parameter which actually means how many cores you require from the whole cluster.
(I quote from - http://spark.apache.org/docs/latest/spark-standalone.html - : "You can also pass an option --total-executor-cores <numCores> to control the number of cores that spark-shell uses on the cluster.")

So what happened when we requested the next settings, an application that wants 12 cores, with 12GB of memory per executor (via "--executor-memory")?
You would expect that the reasonable outcome would be that the scheduler will take 3 workers, each with 4 cores, and each with 12GB, and would run all 12 cores in a dense way...
But that was not the situation, what actually happened that it had spread out all of the wanted cores throughout the whole cluster, and caught all of the machines, 
so we had a core per slave, and 2 slave with 2 cores.
The biggest problem was that on each "Executor" it caught up 12GB of RAM, and the no other applications could run because maybe the "cores" were free, but there was no RAM memory left for more applications.

We had to find a workaround, so we came up with the next solution, we've changed the default settings in the file "$SPARK_HOME/conf/spark-env.sh" in the next parameters:
SPARK_WORKER_INSTANCES (defaults to 1) -> 4
SPARK_WORKER_CORES (default to 4) -> 1

So instead of looking a slave (single physical machine) as a worker, we switched it to having 4 workers with 1 core (CPU) per worker, which gave us the next architecture per slave - Let's call it "Hacked":

And the new cluster topology looks like this:

We no longer look at the cluster as a bunch of slave machines, that each one of them is a worker with multiple cores, but, we look at the cluster as a bunch of workers, each with 1 core, and a certain amount of "worker" memory (In our case we defined it to being 7GB per worker, in the parameter called "spark.executor.memory" in the config file we are passing to "spark-submit" via "--config-file" option).

And for those asking about the parameter you can pass that is called: "spark.deploy.spreadOut" and it's default it true, and set it to false, it will try to stack up all of the wanted cores into a single worker! :)

To conclude our solution, we "Hacked" the cluster to being just a bunch of cores, and we request for each application a certain amount of cores, and the memory per core would be what we defined in the config file.

Thanks to my colleague Nizan Grauer, for the assistance of the creative solution and a lot of playing around with configuration of the cluster.

I know that a lot of people found some kinds of workarounds to this problem, 
But I hope this one helps at least one more person to save time!

If i can assist with anything else, please comment below and I'll try to answer :)

Write Batch Size Error - spark-cassandra-connector

The Use Case:

The Data is currently on Amazon’s S3 storage system and most it is time series data.

We compute and analyze our data using an “Apache Spark” cluster.
While trying to migrate the data to a more efficient storage model (both reads and writes), we tried out Apache Cassandra Database, and during running the first benchmark, we encountered a lot of failures during the “write” part of the Spark nodes.

I'm using the DataStax Java API spark-cassandra-connector.

(You can find some old but useful examples in the DataStax Blog and in the DataStax JavaAPI Documentation)

I started off with using a 2 node Cassandra Cluster, running on 2 c3.2xlarge machines on AWS EC2. I used the DataStax Community AMI to create the nodes and connected two of them into a cluster.

I’ve written a Spark Java application that reads the data from S3 and writes out the RDDs to the Cassandra cluster.

During the “runJob at RDDFunctions.scala:24” stage there were a lot of failures like:

“java.io.IOException: Failed to write 273 batches to test.some_cassandra_table.

……”

Finally the Spark application would fail because of the failures, so i tried to find out the cause of the problem, i went to one of the Cassandra nodes and checked the nodes log located at:

“/var/log/cassandra/system.log”, and many warning messages of the next type coming in all the time:

“

WARN  [SharedPool-Worker-132] 2015-03-19 07:44:43,229 BatchStatement.java:243 - Batch of prepared statements for [test.some_cassandra_table] is of size 5264, exceeding specified threshold of 5120 by 144.

“

After looking for the solution to my problem on Google and coming up with nothing except a variation of the next StackOverFlow answer in many sites, that did not help much, i started looking for the meaning of the Log messages.

I found the definition of the “Batch size threshold” in the Cassandra nodes at  “/etc/cassandra/cassandra.yaml”:

batch_size_warn_threshold_in_kb: 5

Ending up to 5120 bytes, and that’s the value in the logs.

The next step was trying to figure out how to change the write batch size of the DataStax cassandra-spark-driver, and i found the next documentation reference: Link

In the “Tuning” paragraph in mentioned the “spark.cassandra.output.batch.size.rows” parameter you can set to the SparkConf while creating the JavaSparkContext, and i changed it to 5120, but it didn’t give the wanted effect, it multiplied the batch size to being much higher.

It’s default is “auto”, and the “auto”s outcome was much better.

So, after being frustrated with the outcome, i went on reading the source code of the cassandra driver, in the scala class: “WriteConf.scala” and found the usage of another parameter that really made a real change, when using the “auto” default the WriteConf goes straight to the second parameters, “spark.cassandra.output.batch.size.bytes”, value.

I changed the value to being 8192 (instead of the deafult: 1024 * 16 = 16384), making the batch size smaller, and things started working fine.

Although the log message is a warning, it was working some of the time, and sometimes failing, but with changing the parameter, it did not fail anymore.

And if you are asking why the threshold of 5K, i found the next explanation: Link.

Quote: Key reasoning for the desire comes from Patrick McFadden

"Yes that was in bytes. Just in my own experience, I don't recommend more

than ~100 mutations per batch. Doing some quick math I came up with 5k as

100 x 50 byte mutations.

Totally up for debate."

So as we see we can tune these things a bit more, depends on the queries we want to run and depends on the actual environment.

In the future i’ll play a bit more with these configuration, but for now it’s enough for me to continue the benchmark with no errors :)

I felt that someone else must of had the same problem but i didn’t find anything about it written in the internet,

If anyone has a better solution or some other insights i would love to hear, please comment to this post.

I hope this might help someone else.