Distributed Bytes

java

Introduction to Java Profilers

5/08/2019

java

How to tune Garbage Collection in Java

4/13/2019

statistics

What is the p-value? (In layman terms)

9/03/2018

For the stats novice like me understanding what the p-value is can be difficult. This is because when asked, professional statisticians tend to try to give a complete and accurate description of what the p-value is and how it is derived. For example, here is the definition from the American Statistical Association

In statistical hypothesis testing, the p-value or probability value or asymptotic significance is the probability for a given statistical model that, when the null hypothesis is true, the statistical summary (such as the sample mean difference between two compared groups) would be the same as or of greater magnitude than the actual observed results.

I also have heard descriptions that start with a example of a coin that is flipped 1000 times. At that point I go; "buckle up Tim, it's going to be a bumpy ride".

While these are very accurate descriptions of the p-value, as an engineer looking in from the outside into the stats world, I just want a simple definition that gives me some intuition as to what I'm looking at when I see a reported p-value.

So here we go:

To understand what the p-value is, you first need to understand what a null hypothesis is. When running a hypothesis test / experiment, the null hypothesis says that there is no difference or no change between the two tests. The alternate hypothesis is the opposite of the null hypothesis and states that there is a difference between the two tests. The goal of the experiment is usually to disprove the null hypothesis, and to prove/test the alternate hypothesis. Let me illustrate this with some examples:
If you are trying to test whether a new marketing campaign generates more revenue, the null hypothesis is that there is no change in the revenue as a result of the new marketing campaign. And the alternate hypothesis is that the new marketing campaign performs better (or worse) than the previous campaign. If you are trying to prove that a new drug lowers cholesterol, the null hypothesis states that there is no difference in cholesterol between the group with the drug and the group without. While the alternate hypothesis states that the new drug does have an effect on cholesterol levels. If you are trying to test whether a new server version has better or worse performance than the previous version, the null hypothesis is that both server versions have equal performance. And the alternate hypothesis is that there is a meaningful difference in the performance of the old and new server.

So what is the simple layman's definition of the p-value? The p-value is the probability that the null hypothesis is true. That's it! A high p-value indicates that the null hypothesis is likely true and a low p-value indicates that the null hypothesis is likely false.

In the example where we are trying to test whether a new marketing campaign generates more revenue; we can think of the p-value as the probability that the null hypothesis, which states that there is no change in the revenue as a result of the new marketing campaign, is true. If the value of the p-value is 0.25, then there is a 25% probability that there is no real increase or decrease in revenue as a result of the new marketing campaign. If the value of the p-value is 0.04 then there is a 4% probability that there is no real increase or decrease in revenue as a result of the new marketing campaign. As you can surmise, the lower the p-value, the more confident we are that the alternate hypothesis is true, which in this case means that the new marketing campaign causes an increase or decrease in revenue.

So what do p-values really tell us? p-values tell us whether an observation is as a result of a change that was made or is a result of random occurrences. In order to accept a test result we want the p-value to be low. How low you may ask? Well that depends on what standard you want to set / follow. In most fields acceptable p-values should be under 0.05 while in other fields a p-value of under 0.01 is required. This cut-off number is known in statistics as the alpha and results from experiments with p-values below this threshold are considered to be statistically significant. So when a result has a p-value of 0.05 or lower we can say that we are 95% confident that there is an actual difference between the two observations as opposed to just differences due to random variations. And as a result, we have reasonable grounds to support the alternate hypothesis and reject the null hypothesis.

concurrency

Deep dive into the Java Executor Service

12/11/2017

The Java ExecutorService is a construct that allows you to pass a task to be executed by a thread asynchronously. The executor service creates and maintains a reusable pool of threads for executing submitted tasks. The service also manages a queue, which is used when there are more tasks than the number of threads in the pool and there is a need to queue up tasks until there is a free thread available to execute the task.

In this article, we'll focus on the ThreadPoolExecutor implementation of the ExecutorService interface. There are two ways to instantiate a Thread Pool Executor, you can either directly instantiate it using one of its constructor overloads or you can use one of the factory methods in the Executors class.

ExecutorService executorService = 
          new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS,
          new LinkedBlockingQueue<Runnable>());

Directly instantiating a ThreadPoolExecutor with 10 threads, a keepAliveTime of 0 milliseconds and a LinkedBlockingQueue

ExecutorService executor = Executors.newFixedThreadPool(10);

Instantiating a ThreadPoolExecutor with 10 threads using an Executors factory method

ExecutorService executor = Executors.newSingleThreadExecutor();

Instantiating a ThreadPoolExecutor with a single thread using an Executors factory method

ExecutorService executor = Executors.newCachedThreadPool();

Instantiating a ThreadPoolExecutor which adds threads to the pool as needed using an Executors factory method

When you instantiate your Executor Service a few parameters are initialized. Depending on how you instantiated your Executor Service you may manually specify these parameters or they may be provided for you by default. These parameters are:

corePool size
maxPool size
workQueue
keepAliveTime
threadFactory
rejectedExecutionHandler

Using one of the factory methods available in the Executors class simply selects some default values for the above for you based on your inputs. Executors.newSingleThreadExecutor() creates a pool with a core size of 1, max size of 1, a keepAliveTime of 0ms (which means that the thread in the pool would stay alive unless explicitly closed), an unbounded LinkedBlockingQueue, the default threadFactory and the default rejectedExecutionHandler. While Executors.newFixedThreadPool(10) creates a pool with a core size of 10, max size of 10, a keepAliveTime of 0ms, an unbounded LinkedBlockingQueue, the default threadFactory and the default rejectedExecutionHandler.

So what do all these parameters mean?

The core pool size is the minimum number of threads that should be kept in the pool. The number of threads may grow to reach the max pool size (if it is higher than the core pool size), but in general, it represents the number of threads you expect to have alive in the pool. When a task is submitted to the executor, it checks if the actual running number of threads is less than the core pool size and if it is then it creates a new worker using the specified threadFactory.

The max pool size is the maximum number of workers that can be in the pool. If the max pool size is greater than the core pool size, it means that the pool can grow in size i.e. more workers can be added to the pool. Workers are added to the pool when a task is submitted but the work queue is full. Every time this happens, a new worker is added until the max pool size is reached. If the max pool size has already been reached and the work queue is full then the next task will be rejected.

The work queue is used to queue up tasks for the available worker threads. The queue can be bounded or unbounded. For bounded queues, setting the queue size is an important exercise as it affects how the worker pool grows and when you start running into RejectedExecutionExceptions. If you have a work pool that you expect to grow; say from a core pool size of 20 workers to a max of 100 workers, then you may not want to set the queue size to a number that is too high like 10,000 because it means that 10,000 tasks must be enqueued before the each additional worker gets added to the pool. Unbounded queues and bounded queues with very high capacities are more suited to be used with fixed size pools (i.e. pools where the core and max pools sizes are the same).

If a thread pool grows to the max size, how does it shrink back to the core size? That's where the keepAliveTime comes in. If the current number of worker threads exceeds the core pool size and a keepAliveTime is set then worker threads are shut down when there is no more work to do until the number of worker threads is back to the core pool size; a thread will wait for work for the keepAlive time, and when that is exceeded and no work arrives, it will shut down.

Side Note 1: You can set allowCoreThreadTimeOut to true on your ThreadPoolExecutor instance, and if you do so then not only workers threads that exceed the core pool size get shutdown on idle, but also core worker threads also get shutdown on idle. By default this is set to false.

Side Note 2: If your worker threads acquire and maintain expensize resources and only release those resources on shutdown, then it becomes important to optimally configure your keepAlive time. A keepAlive time of 0 ms means that your workers never shutdown after they are created, unless the executor service itself is shut down.

Most times using the Default ThreadFactory is sufficient. The default thread factory creates worker threads that have a normal priority and are not daemon threads. It also gives the threads a name with the format: "pool-{poolNumber}-thread-{threadNumber}". If you want to customize any of these attributes such as the thread name or priority then you should provide your own ThreadFactory implementation. Another benefit of providing your own thread factory implementation is that you can set the thread's uncaught exception handler which can be very useful in combating silent failures.

Speaking of exceptions, I mentioned that for pools with a bounded work queue, task rejections occur when the queue is full and no more workers can be added. You can configure a handler to run when such a rejection occurs. These handlers are called "Policies". By default, the AbortPolicy is used, which throws a RejectedExecutionException. You can choose to use another Policy such as the DiscardPolicy, which simply discards the task silently, the CallerRunsPolicy, which executes the task on the calling thread instead of one of the worker threads, or any another policy implementation you create.

To wrap up, the Java Executor Service hides a lot of complexity but also makes it easy for you to dive in and tweak the inner workings if you so choose. The Executors class provides a lot of factory methods that address different use cases; `newFixedThreadPool()` for when you just need a fixed number of threads that execute tasks, `newCachedThreadPool()` for when you want to create new threads as needed and shrink the pool when not needed, etc. In many cases, these pre-defined pools would meet your needs. However, if you have more defined parameters, then it helps to know some of the knobs you can tweak and how that affects the thread pool's behavior.

concurrency

A Quick Primer on Python Concurrency

10/26/2017

Python is often thought of as a single threaded language but there are several avenues for executing tasks concurrently.
The threading module allows us to spin up native operating system threads to execute multiple tasks concurrently. The threading API has methods for creating thread objects and then using the object to start and join on the underlying thread.


# define the function to execute in a thread
def do_some_work(val):
    print ("doing some work in thread") 
    print ("echo: {}".format(val))
    return 

val = "text" 
# create thread object passing in the target function and optional args in constructor
t = threading.Thread(target=do_some_work, args=(val,)) 
# start the thread
t.start()
# block execution of the main thread until thread t is completed
t.join()

The threading module also provides several synchronization and inter-thread communication mechanisms for when threads need to communicate and coordinate with each other, or for when multiple threads are mutating the same area of memory. Locks and Queues are the most common of those Synchronization methods but Python also provides RLocks, Semaphores, Conditions, Events and Barrier implementations in the threading API.


lock = threading.Lock()
### assume that code below runs in multiple threads ###
lock.acquire() # acquire the lock preventing other threads from doing so
try:
     # access shared resource
finally:
        lock.release() # release the lock so that other blocked threads can now run


queue = Queue()
## assume the code below runs in a separate thread t1 ###
def producer(queue):
        item = make_an_item()
        queue.put(item)
## assume the code below runs in a separate thread t2 ###
def consumer(queue):
        item = queue.get() #gets item put in the queue by another thread. Blocks if item not there yet
        queue.task_done() # marks the last item retrieved as done

However, the current implementation of cpython has a Global Interpreter Lock (GIL) to make Python easier to implement and faster to run for single threaded programs. But as a result of the GIL, which only allows one thread to run at a time, Threading is not suitable for CPU-bound tasks (tasks in which most of the time is spent performing a computation instead of waiting on IO). So instead we have the multiprocessing package. The multiprocessing package uses processes instead of threads as the actors of parallel execution. And the multiprocessing API tries to mimic the threading API as much as possible, to reduce the amount of dissonance between the two and to make switching easier.


# define the function to execute in a new process
def generate_hash(text):
      return hashlib.sha384(text).hexdigest()
text = "some long text here…"
if __name__ == '__main__':
    # create process object passing in the target function and optional args in constructor
    p = multiprocessing.Process(target= generate_hash,args=(text,))
       # start and join the process
       p.start()
       p.join()

One of the major areas where there is a difference between the threading and multiprocessing APIs is in the implementation of shared state. Threads automatically share memory with each other but processes don't. So, special accommodations must be made to allow processes to communicate and share state. Processes can either allocate and use OS shared memory areas, or can communicate with a server process which maintains shared data.
The concurrent.futures module provides a layer of abstraction over both concurrency mechanisms (threads and processes).
It was also the introduction of Futures into Python. In Python, a future represents a pending result and it also allows us to manage the execution of the computation that produces the result. Future API methods include result(), cancel() and add_done_callback(fn)


# define the function to execute in a new process
def generate_hash(text):
      return hashlib.sha384(text).hexdigest()
text = b"some long text here..."
executor = ProcessPoolExecutor() # can be replaced with `ThreadpoolExecutor()`
future_result = executor.submit(generate_hash, text) # submit a job to the pool, immediately returns a future object

Finally, the most recent addition to the Python concurrency family is the asyncio module. asyncio brings single threaded asynchronous programming to Python. It provides an event loop which runs specialized functions called coroutines, A coroutine has the ability to pause itself and yield control back to the event loop when it needs to wait for IO or some other long running task. The event loop can then go on and execute other coroutines and resume the prior coroutine when an event occurs that indicates that the IO or long running task is complete. As a result, we have multiple tasks running on the same thread and yielding to one another instead of blocking.


# a coroutine function as denoted by the async keyword
async def delayed_hello():
        print("Hello ")
       # the coroutine will pause here and yield back to the event loop
    await asyncio.sleep(1)
      print("World!")
# get the event loop
loop = asyncio.get_event_loop()
# pass the coroutine to the event loop for execution
loop.run_until_complete(delayed_hello())
loop.close()

There are several resources that provided an in-depth look into Python concurrency like the Python Module of the Week blog and the Python Parallel Programming Cookbook. If you are a Pluralsight user, you can also checkout my Pluralsight course Python Concurrency: Getting Started.

analytics

Step by Step Guide to Installing and Configuring Spark 2.0 to Connect with Cassandra 3.x

1/06/2017

In this guide, we will be installing Scala 2.11, Spark 2.0 as a service, and the DataStax spark-cassandra-connector library on the client program. If you have any of these software packages installed and configured already you can skip that step. This guide assumes you have a Cassandra 3.x cluster that is already up and running. For more information on installing and using Cassandra visit http://cassandra.apache.org/.Note: The following steps should be performed on all the nodes in the cluster unless otherwise noted.

Install Scala 2.11

Ensure you have Java installed


$ java -version
java version "1.8.0_91"
Java(TM) SE Runtime Environment (build 1.8.0_91-b14)
Java HotSpot(TM) 64-Bit Server VM (build 25.91-b14, mixed mode)

If you don't have Java installed follow this tutorial to get Java 8 installed.
Install Scala 2.11.8


$ wget www.scala-lang.org/files/archive/scala-2.11.8.deb
$ sudo dpkg -i scala-2.11.8.deb
$ scala -version

Install SBT 0.13


$ echo "deb https://dl.bintray.com/sbt/debian /" | sudo tee -a /etc/apt/sources.list.d/sbt.list
$ sudo apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 2EE0EA64E40A89B84B2DF73499E82A75642AC823
$ sudo apt-get update
$ sudo apt-get install sbt

Install Spark 2.0

Download Spark 2.0 from https://spark.apache.org/downloads.html and unpack the tar file


$ wget http://d3kbcqa49mib13.cloudfront.net/spark-2.0.2-bin-hadoop2.7.tgz
$ tar zxf spark-2.0.2-bin-hadoop2.7.tgz
$ sudo mv spark-2.0.2-bin-hadoop2.7/ /usr/local/spark/

Update system variables


$ sudo nano /etc/environment

Add an environment variable called SPARK_HOME


export SPARK_HOME=/usr/local/spark

At the end of the PATH variable add $SPARK_HOME/bin


PATH="<previous_entries>:/usr/local/spark/bin"

Refresh the environment


$ source /etc/environment

Create spark user and make it the owner of the SPARK_HOME directory


$ sudo adduser spark --system --home /usr/local/spark/ --disabled-password
$ sudo chown -R spark:root /usr/local/spark

Create Log and Pid directories


$ sudo mkdir /var/log/spark
$ sudo chown spark:root /var/log/spark
$ sudo -u spark mkdir $SPARK_HOME/run

Create the Spark Configuration files

Create the Spark Configuration files by copying the templates


$ sudo cp /usr/local/spark/conf/spark-env.sh.template /usr/local/spark/conf/spark-env.sh
$ sudo cp /usr/local/spark/conf/spark-defaults.conf.template /usr/local/spark/conf/spark-defaults.conf
$ sudo chown spark:root /usr/local/spark/conf/spark-*

Edit the Spark Environment file spark-env.sh


export SPARK_LOG_DIR=/var/log/spark
export SPARK_PID_DIR=${SPARK_HOME}/run

Configure Spark nodes to join cluster

If you will not be managing Spark using the Mesos or YARN cluster managers, you'll be running Spark in what is called "Standalone Mode".
In standalone mode, Spark will have a master node (which is the cluster manager) and worker nodes. You should select one of the nodes in your cluster to be the master. Then on every worker node you must edit the /etc/spark/conf/spark-env.sh to point to the host where the Spark Master runs.


# Options for the daemons used in the standalone deploy mode
export SPARK_MASTER_HOST=<spark_master_ip_or_hostname_here>

You can also change other elements of the default configuration by editing the /etc/spark/conf/spark-env.sh. Some other configs to consider are:

SPARK_MASTER_PORT / SPARK_MASTER_WEBUI_PORT, to use non-default ports
SPARK_WORKER_CORES, to set the number of cores to use on this machine
SPARK_WORKER_MEMORY, to set how much memory to use (for example 1000MB, 2GB)
SPARK_WORKER_PORT / SPARK_WORKER_WEBUI_PORT
SPARK_WORKER_INSTANCE, to set the number of worker processes per node
SPARK_WORKER_DIR, to set the working directory of worker processes

Installing Spark as a service

Run the following commands to create a service for the spark-master and spark-worker


$ sudo cp /etc/init.d/skeleton /etc/init.d/spark-master
$ sudo chmod 0755 /etc/init.d/spark-master
$ sudo cp /etc/init.d/skeleton /etc/init.d/spark-worker
$ sudo chmod 0755 /etc/init.d/spark-worker
$ sudo update-rc.d spark-master defaults 99
$ sudo update-rc.d spark-worker defaults 99

Edit the /etc/init.d/spark-worker file. If a variable or function already exists then replace it with the text below.


DESC="Spark Worker"
NAME=spark-worker
SPARK_HOME=/usr/local/spark
PIDFILE=$SPARK_HOME/run/spark--org.apache.spark.deploy.worker.Worker-1.pid
export SPARK_HOME


# Exit if the package is not installed
#[ -x "$DAEMON" ] || exit 0


if [ -f $SPARK_HOME/conf/spark-env.sh ];then
        . $SPARK_HOME/conf/spark-env.sh
else
        echo "$SPARK_HOME/conf/spark-env.sh not found. Cannot start service."
fi


#
# Function that returns 0 if process is running, or nonzero if not.
#
# The nonzero value is 3 if the process is simply not running, and 1 if the
# process is not running but the pidfile exists (to match the exit codes for
# the "status" command; see LSB core spec 3.1, section 20.2)
#
is_running()
{
    CMD_PATT="org.apache.spark.deploy.worker.Worker"
    if [ -f $PIDFILE ]; then
        pid=`cat $PIDFILE`
        grep -Eq "$CMD_PATT" "/proc/$pid/cmdline" 2>/dev/null && return 0
        return 1
    fi
    return 3
}


#
# Function that starts the daemon/service
#
do_start()
{
        # Return
        #   0 if daemon has been started
        #   1 if daemon was already running
        #   2 if daemon could not be started


        [ -e `dirname "$PIDFILE"` ] || \
                install -d -ospark -groot -m755 `dirname $PIDFILE`


        start-stop-daemon --start --quiet --chuid spark --pidfile $PIDFILE  \
                --exec $SPARK_HOME/sbin/start-slave.sh  \
                --test > /dev/null \
                || return 1
        start-stop-daemon --start --quiet --chuid spark --pidfile $PIDFILE \
                --exec $SPARK_HOME/sbin/start-slave.sh -- spark://$SPARK_MASTER_IP:$SPARK_MASTER_PORT \
                || return 2


}


#
# Function that stops the daemon/service
#
do_stop()
{
        start-stop-daemon --stop --quiet --retry=TERM/30/KILL/5 --pidfile $PIDFILE
        RETVAL="$?"
        rm -f $PIDFILE
        return "$RETVAL"
}


#
# Function that sends a SIGHUP to the daemon/service
#
do_reload() {
        #
        # If the daemon can reload its configuration without
        # restarting (for example, when it is sent a SIGHUP),
        # then implement that here.
        #
        start-stop-daemon --stop --signal 1 --quiet --pidfile $PIDFILE
        return 0
}


...
status)
      is_running
      stat=$?
      case "$stat" in
              0) log_success_msg "$DESC is running" ;;
              1) log_failure_msg "could not access pidfile for $DESC" ;;
              *) log_success_msg "$DESC is not running" ;;
      esac
      exit "$stat"
      ;;
...

Edit the /etc/init.d/spark-master file. If a variable or function already exists then replace it with the text below.


DESC="Spark Master"
NAME=spark-master
SPARK_HOME=/usr/local/spark
PIDFILE=$SPARK_HOME/run/spark--org.apache.spark.deploy.master.Master-1.pid
export SPARK_HOME


# Exit if the package is not installed
#[ -x "$DAEMON" ] || exit 0


if [ -f $SPARK_HOME/conf/spark-env.sh ];then
        . $SPARK_HOME/conf/spark-env.sh
else
        echo "$SPARK_HOME/conf/spark-env.sh not found. Cannot start service."
fi

#
# Function that returns 0 if process is running, or nonzero if not.
#
# The nonzero value is 3 if the process is simply not running, and 1 if the
# process is not running but the pidfile exists (to match the exit codes for
# the "status" command; see LSB core spec 3.1, section 20.2)
#
is_running()
{
    CMD_PATT="org.apache.spark.deploy.master.Master"
    if [ -f $PIDFILE ]; then
        pid=`cat $PIDFILE`
        grep -Eq "$CMD_PATT" "/proc/$pid/cmdline" 2>/dev/null && return 0
        return 1
    fi
    return 3
}


#
# Function that starts the daemon/service
#
do_start()
{
        # Return
        #   0 if daemon has been started
        #   1 if daemon was already running
        #   2 if daemon could not be started


        [ -e `dirname "$PIDFILE"` ] || \
                install -d -ospark -groot -m755 `dirname $PIDFILE`


        start-stop-daemon --start --quiet --chuid spark --pidfile $PIDFILE --exec $SPARK_HOME/sbin/start-master.sh --test > /$
                || return 1
        start-stop-daemon --start --quiet --chuid spark --pidfile $PIDFILE --exec $SPARK_HOME/sbin/start-master.sh  \
                || return 2
}


#
# Function that stops the daemon/service
#
do_stop()
{
        start-stop-daemon --stop --quiet --retry=TERM/30/KILL/5 --pidfile $PIDFILE


        RETVAL="$?"
        rm -f $PIDFILE
        return "$RETVAL"
}


#
# Function that sends a SIGHUP to the daemon/service
#
do_reload() {
        #
        # If the daemon can reload its configuration without
        # restarting (for example, when it is sent a SIGHUP),
        # then implement that here.
        #
        start-stop-daemon --stop --signal 1 --quiet --pidfile $PIDFILE
        return 0
}


...
status)
      is_running
      stat=$?
      case "$stat" in
              0) log_success_msg "$DESC is running" ;;
              1) log_failure_msg "could not access pidfile for $DESC" ;;
              *) log_success_msg "$DESC is not running" ;;
      esac
      exit "$stat"
      ;;
...

Running Spark as a service

Start the Spark master node first. On whichever node you've selected to be master, run:


$ sudo service spark-master start

On all the other nodes, start the workers:


$ sudo service spark-worker start

To stop Spark, run the following commands on the appropriate nodes


$ sudo service spark-worker stop
$ sudo service spark-master stop

Service logs will be stored in /var/log/spark.

Testing the Spark service

To test the Spark service, start spark-shell on one of the nodes.


$ spark-shell --master spark://<IP>:<Port>

When the prompt comes up, execute the following line of code:


$ scala> sc.parallelize( 1 to 1000 ).sum()

Get Spark-Cassandra-Connector on the client

The spark-cassandra-connector is a Scala library that exposes Cassandra tables as Spark RDDs, lets you write Spark RDDs to Cassandra tables, and allows you to execute arbitrary computations and CQL queries that are distributed to the Cassandra nodes holding the data, which allows them to be fast. Your code + the spark-cassandra-connector and all dependencies are packaged up and sent to the Spark nodes.
If you are writing ad-hoc queries / computations from the spark-shell. Start up the shell by running the command:


$ spark-shell --conf spark.cassandra.connection.host=<master-ip-address> --packages datastax:spark-cassandra-connector:2.0.0-M2-s_2.11

The --packages option downloads the connector and all of its dependencies from the Spark Packages site and places it in the path of the Spark Driver and all Spark Executors.
If you are writing a Scala application; configure a new Scala project. Your build.sbt file should look something like this:


name := "MySparkProject"


version := "1.0"


scalaVersion := "2.11.8"


val sparkVersion = "2.0.2"


resolvers += "Spark Packages Repo" at "https://dl.bintray.com/spark-packages/maven"


libraryDependencies ++= Seq(
  "org.apache.spark"      % "spark-core_2.11"  % sparkVersion,
  "org.apache.spark"      % "spark-sql_2.11"   % sparkVersion,
  "datastax"              % "spark-cassandra-connector" % "2.0.0-M2-s_2.11"
)

Testing the connector

To start out, create a simple keyspace and table in Cassandra. Run the following statements in cqlsh:


CREATE KEYSPACE test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1 };
CREATE TABLE test.kv(key text PRIMARY KEY, value int);

Then insert some example data:


INSERT INTO test.kv(key, value) VALUES ('key1', 1);
INSERT INTO test.kv(key, value) VALUES ('key2', 2);

For this test, we'll use the spark-shell.


$ spark-shell --conf spark.cassandra.connection.host=<master-ip-address> --packages datastax:spark-cassandra-connector:2.0.0-M2-s_2.11


// import the spark connector namespace
import com.datastax.spark.connector._


// read Cassandra data into an RDD
val rdd = sc.cassandraTable("test", "kv")
println(rdd.count)
println(rdd.first)
println(rdd.map(_.getInt("value")).sum)  


// Add two more rows to the table
val collection = sc.parallelize(Seq(("key3", 3), ("key4", 4)))
collection.saveToCassandra("test", "kv", SomeColumns("key", "value"))

Introduction to Java Profilers

How Java Profilers Work

Performance Analysis using Java Profilers

How to tune Garbage Collection in Java

Sizing the heap

Tuning GC performance

Fixing Concurrent Mode Failures