Spark Cassandra Integration with spark-cassandra-connector

In here, I am going to show how to integrate local single node Cassandra db with standalone spark using spark-cassandra-connector.

Setup Cassandra, Spark, Scala & ScalaBuildTool

1. Download Cassandra & Spark. I am using Cassandra version 3.11.2  and Spark version 2.2.1 .

http://cassandra.apache.org/download/
https://spark.apache.org/releases/spark-release-2-2-1.html
https://www.scala-lang.org/download/2.11.8.html
https://www.scala-sbt.org/download.html

2. Environment setup in .profile

#cassandra setup
export CASSANDRA_HOME=/home/dhanuka/software/apache-cassandra-3.11.2

#spark, sbt and scala setup
export SPARK_HOME=/home/dhanuka/software/spark/spark-2.2.1-bin-hadoop2.7
export SBT_HOME=/home/dhanuka/software/spark/sbt-launcher-packaging-0.13.13
export SCALA_HOME=/home/dhanuka/software/scala-2.11.8

PATH=$PATH:$JAVA_HOME/bin:$MAVEN_HOME/bin:SPARK_HOME/bin:$SBT_HOME/bin:$SCALA_HOME/bin:CASSANDRA_HOME/bin

3. $ source ~/.profile 

Create Cassandra Keyspace and Table

 1. Start cassandra with following command

$ cassandra -f

2. Start CQL shell

$ cqlsh

3. Create keyspace and a table

cqlsh> CREATE KEYSPACE people WITH replication = {'class': 'SimpleStrategy', 'replication_factor':1};

cqlsh> use people;

cqlsh:people> CREATE TABLE users(
          ... id varchar ,
          ... first_name varchar,
          ... last_name varchar,
          ... city varchar,
          ... emails varchar,
          ... PRIMARY KEY (id));
 
cqlsh:people>  Insert into users (id,first_name,last_name,city,emails) values('1','dhanuka','ranasinghe','colombo','dhanuka.priyanath@gmail.com');

 cqlsh:people> select * from users;

 id      | city    | emails                      | first_name | last_name
---------+---------+-----------------------------+------------+------------
 1 | colombo | dhanuka.priyanath@gmail.com |    dhanuka | ranasinghe




Build spark-cassandra-connector.

1. clone from git hub repository.

$  git clone https://github.com/datastax/spark-cassandra-connector.git 

$  cd spark-cassandra-connector

2. Build the project with scala 2.11 and cassandra 3.11.2

$  spark-cassandra-connector$ sbt -Dscala-2.11=true -Dtest.cassandra.version=3.11.2 assembly

You can find the jar location below.

$  spark-cassandra-connector/spark-cassandra-connector/target/full/scala-2.11/spark-cassandra-connector-assembly-2.0.7-82-g0369a7b.jar

$  mv  spark-cassandra-connector-assembly-2.0.7-82-g0369a7b.jar   spark-cassandra-connector-assembly-2.0.7.jar


 Connect Spark with Cassandra through Spark-Shell

1. Copy cassandra-connector-assembly-2.0.7.jar to spark jars location. Copy to below location

$  cp cassandra-connector-assembly-2.0.7.jar  $SPARK_HOME/jars

2.  Start spark-shell

$ spark-shell --jars $SPARK_HOME/jars/spark-cassandra-connector-assembly-2.0.7.jar

3. Stop current spark context

scala> sc.stop

4. Program to read Cassandra from spark

scala> import com.datastax.spark.connector._, org.apache.spark.SparkContext, org.apache.spark.SparkContext._, org.apache.spark.SparkConf
import com.datastax.spark.connector._
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf


scala> val conf = new SparkConf(true).set("spark.cassandra.connection.host", "localhost")
conf: org.apache.spark.SparkConf = org.apache.spark.SparkConf@2c2a7d53

scala> val sc = new SparkContext(conf)
sc: org.apache.spark.SparkContext = org.apache.spark.SparkContext@15914bb5

scala> val test_spark_rdd = sc.cassandraTable("people", "users")
test_spark_rdd: com.datastax.spark.connector.rdd.CassandraTableScanRDD[com.datastax.spark.connector.CassandraRow] = CassandraTableScanRDD[0] at RDD at CassandraRDD.scala:19

scala> test_spark_rdd.first
res1: com.datastax.spark.connector.CassandraRow = CassandraRow{id: 1, city: colombo, emails: dhanuka.priyanath@gmail.com, first_name: dhanuka, last_name: ranasinghe}

References:

[1] https://www.datastax.com/dev/blog/kindling-an-introduction-to-spark-with-cassandra-part-1

[2] https://www.youtube.com/watch?v=jpEABn80OCU

Setup VNode Cassandra cluster in local machine

Here I am using Ubuntu Linux machine and also tool called Cassandra Cluster Manager (ccm).

Steps

1. :~/opensource$ mkdir cassandra

2. :~/opensource$ cd cassandra/

3. :~/opensource/cassandra$ git clone https://github.com/riptano/ccm.git

4. :~/opensource/cassandra$ cd ccm/

5. :~/opensource/cassandra/ccm$ ls

ccm  ccmlib  license.txt  MANIFEST.in  misc  NETWORK_ALIASES.md  README.md  setup.py  ssl  tests

6. :~/opensource/cassandra/ccm$ sudo ./setup.py install

7. :~/opensource/cassandra/ccm$ ccm create -v 3.11.2 -n 3 my_cluster --vnodes

10:22:00,960 ccm INFO Downloading http://archive.apache.org/dist/cassandra/3.11.2/apache-cassandra-3.11.2-bin.tar.gz to /tmp/ccm-Wo1DBu.tar.gz (36.656MB)
  38436262  [100.00%]10:24:09,353 ccm INFO Extracting /tmp/ccm-Wo1DBu.tar.gz as version 3.11.2 ...
Current cluster is now: my_cluster

cluster name: my_cluster
cassandra version: 3.11.2
number of nodes: 3

8.  :~/opensource/cassandra/ccm$ ccm list

*my_cluster

9. :~/opensource/cassandra/ccm$ ccm status
Cluster: 'my_cluster'
---------------------
node1: DOWN (Not initialized)
node3: DOWN (Not initialized)
node2: DOWN (Not initialized)

10. :~/opensource/cassandra/ccm$ ccm start
      :~/opensource/cassandra/ccm$ ccm status
Cluster: 'my_cluster'
---------------------
node1: UP
node3: UP
node2: UP


 11. :~/opensource/cassandra/ccm$ ccm node1 status

Datacenter: datacenter1
=======================
Status=Up/Down
|/ State=Normal/Leaving/Joining/Moving
--  Address    Load       Tokens       Owns (effective)  Host ID                               Rack
UN  127.0.0.1  88.93 KiB  256          64.1%             5e9d0ad1-d7d0-4023-b71c-2f57375c2573  rack1
UN  127.0.0.2  69.9 KiB   256          67.3%             e8ea9b63-7846-4494-bd0d-2c8dfcb9274b  rack1
UN  127.0.0.3  103.62 KiB  256          68.7%             301887a5-b386-47c6-8a59-3f71ea09b9fc  rack1


12. :~/opensource/cassandra/ccm$ ccm node1 ring


Datacenter: datacenter1
==========
Address    Rack        Status State   Load            Owns                Token                                      
                                                                          9184122387525172037                        
127.0.0.3  rack1       Up     Normal  103.62 KiB      68.67%              -9188398778489235111                       
127.0.0.2  rack1       Up     Normal  69.9 KiB        67.27%              -9176942363448371450                       
127.0.0.3  rack1       Up     Normal  103.62 KiB      68.67%              -9165278195783829162                       
127.0.0.3  rack1       Up     Normal  103.62 KiB      68.67%              -9154226470359649220                       
127.0.0.3  rack1       Up     Normal  103.62 KiB      68.67%              -9105345187931701041                       
127.0.0.1  rack1       Up     Normal  88.93 KiB       64.06%              -9059861789902146091                       
127.0.0.1  rack1       Up     Normal  88.93 KiB       64.06%              -9006955999491952161                       
127.0.0.1  rack1       Up     Normal  88.93 KiB       64.06%              -8991976771179386553                       
127.0.0.1  rack1       Up     Normal  88.93 KiB       64.06%              -8974379714699893637                       
127.0.0.1  rack1       Up     Normal  88.93 KiB       64.06%              -8964003200974837827                       
127.0.0.2  rack1       Up     Normal  69.9 KiB        67.27%              -8885086536332616726                       
127.0.0.3  rack1       Up     Normal  103.62 KiB      68.67%              -8882333637674950203                       
127.0.0.2  rack1       Up     Normal  69.9 KiB        67.27%              -8866726432815074132     


13. :~/opensource/cassandra/ccm$ ccm stop

14. :~/opensource/cassandra/ccm$ ccm -h

Anatomy of Micro services based & distributed Master Data Management Architecture

Please note this is improved version of previously[1] explained architecture.

[1] http://dhanuka84.blogspot.com/2017/06/master-data-management-for-distributed.html

Master Data Change Detection

1. Admin Panel will change or add master data.
2. Changes goes to DB.
3. Update Hibernate First Level Cache.
4. Distribute data via Kafka.
5. Kafka Consumer detect master data change notification.
6. Update Hazelcast Distributed map with which master data has been changed.
7. Scheduler application task read from HZ map and get to know about which master data changed.
8. Get latest data for changed master data from master data micro service.
9. Update LocalEntityMap with business key and Entity.
10. Update Business key to Primary key map.
11. Removed master data notification record from HZ distributed map.

Scheduler Application Load Balancing

1. Get Business key to primary key mapping to all master data.
2. Update Business key to primary key mapping
3. Read Business key to primary key map then load balanced among HZ cluster members.
4. update HZ distributed map.
5. Keep business keys which load balanced to local application node.

Processing using master data within scheduler task

1. Read from HZ distributed map and get master data local to this node for processing and Keep them for processing.
2. Take business keys to process.
3. Get locally cached master data for business keys for processing.
4. If can’t find in the cache get from micro service.

Simple explanation about basic machine learning algorythms with R

Supervised   

Supervised learning is where you have input variables (x) and an output variable (Y) and you use an algorithm to learn the mapping function from the input to the output.

Y = f(X)

The goal is to approximate the mapping function so well that when you have new input data (x) that you can predict the output variables (Y) for that data.

Regression vs Classification   

Regression is used to predict continuous values. Classification is used to predict which class a data point is part of (discrete value).

                           

Regression

In the case of regression, the target variable is continuous — meaning that it can take any value within a specified range. Input variables, on the other hand, can be either discrete or continuous.

Linear Regression

Math:

SepalLength = a * PetalWidth + b* PetalLength +c

Code:

# Load required packages

library(ggplot2)

# Load iris dataset

data(iris)

# Have a look at the first 10 observations of the dataset

head(iris)

# Fit the regression line

fitted_model <- lm(Sepal.Length ~ Petal.Width + Petal.Length, data = iris)

# Get details about the parameters of the selected model

summary(fitted_model)

# Plot the data points along with the regression line

ggplot(iris, aes(x = Petal.Width, y = Petal.Length, color = Species)) +

 geom_point(alpha = 6/10)  +

 stat_smooth(method = "lm", fill="blue", colour="grey50", size=0.5, alpha = 0.1)

Logistic Regression

The difference is that the regression line is not straight anymore.

Math:

Y=g(a*X1+b*X2)

...where g() is the logistic function.

Code:

# Load required packages

library(ggplot2)

# Load data

data(mtcars)

# Keep a subset of the data features that includes on the measurement we are interested in

cars <- subset(mtcars, select=c(mpg, am, vs))

# Fit the logistic regression line

fitted_model <- glm(am ~ mpg+vs, data=cars, family=binomial(link="logit"))

# Plot the results

ggplot(cars, aes(x=mpg, y=vs, colour = am)) + geom_point(alpha = 6/10) +

 stat_smooth(method="glm",fill="blue", colour="grey50", size=0.5, alpha = 0.1, method.args=list(family="binomial"))

Decision Trees (Classification or Regression)

What they basically do is draw a “map” of all possible paths along with the corresponding result in each case.

Based on a tree like this, the algorithm can decide which path to follow at each step depending on the value of the corresponding criterion.

Code:

# Include required packages

#install.packages("party")

#install.packages("partykit")

library(party)

library(partykit)

# Have a look at the first ten observations of the dataset

print(head(readingSkills))

input.dat <- readingSkills[c(1:105),]

# Grow the decision tree

output.tree <- ctree(

 nativeSpeaker ~ age + shoeSize + score,

 data = input.dat)

# Plot the results

plot(as.simpleparty(output.tree))

Unsupervised

Unsupervised learning is where you only have input data (X) and no corresponding output variables.

The goal for unsupervised learning is to model the underlying structure or distribution in the data in order to learn more about the data.

These are called unsupervised learning because unlike supervised learning above there is no correct answers and there is no teacher. Algorithms are left to their own devises to discover and present the interesting structure in the data.

   

Clustering

With clustering, if we have some initial data at our disposal, we want to form groups so that the data points belonging to some group are similar and are different from data points of the other groups , such as grouping customers by purchasing behavior..

Algorythm :

1. Initialization step: For k=3 clusters, the algorithm randomly selects three points as centroids for each cluster.
2. Cluster assignment step: The algorithm goes through the rest of the data points and assigns each one of them to the closest cluster.
3. Centroid move step: After cluster assignment, the centroid of each cluster is moved to the average of all points belonging to the cluster.

Steps 2 and 3 are repeated multiple times until there is no change to be made regarding cluster assignments.

Code:

# Load required packages

library(ggplot2)

library(datasets)

# Load data

data(iris)

# Set seed to make results reproducible

set.seed(20)

# Implement k-means with 3 clusters

iris_cl <- kmeans(iris[, 3:4], 3, nstart = 20)

iris_cl$cluster <- as.factor(iris_cl$cluster)

# Plot points colored by predicted cluster

ggplot(iris, aes(Petal.Length, Petal.Width, color = iris_cl$cluster)) + geom_point()

Colored by Species

Code:

# Load required packages

library(ggplot2)

library(datasets)

# Load data

data(iris)

# Set seed to make results reproducible

set.seed(20)

# Implement k-means with 3 clusters

iris_cl <- kmeans(iris[, 3:4], 3, nstart = 20)

iris_cl$cluster <- as.factor(iris_cl$cluster)

# Plot points colored by predicted cluster

ggplot(iris, aes(Petal.Length, Petal.Width, color = iris$Species)) + geom_point()

Association

An association rule learning problem is where you want to discover rules that describe large portions of your data, such as people that buy X also tend to buy Y.

References:

https://dzone.com/articles/3-machine-learning-algorithms-you-need-to-know?edition=329502&utm_source=Weekly%20Digest&utm_medium=email&utm_campaign=Weekly%20Digest%202017-10-04

https://machinelearningmastery.com/supervised-and-unsupervised-machine-learning-algorithms/

AB Deployment Pattern to Support Data Replication & Processing

Sending Logs to ELK Stack through Logback

Sample Kibana Dashboard

Pre-Requirements & Steps to Setup

1. Java Application which already configured with Logback as Logging manager and use Groovy to configure Logback.

Sample Java Start up script 

java \
-Xms512m -Xmx1024m \
-XX:+HeapDumpOnOutOfMemoryError \
-XX:HeapDumpPath="/home/uranadh/opensource/kafka_connect_config/heap-dump.hprof" \
-cp "distributed-services-1.0.0.jar:lib/*" \
 -Dlogback.configurationFile=/home/uranadh/opensource/kafka_connect_config/logback.groovy  \
 org.reactor.monitoring.application.internal.Member

2. Configure logback.groovy file with Logstash Appender.

Please note here we use below Logstash  TCP Appender.

https://github.com/logstash/logstash-logback-encoder#pattern-json-provider

Groovy file

import ch.qos.logback.classic.AsyncAppender;
import ch.qos.logback.classic.encoder.PatternLayoutEncoder;
import ch.qos.logback.core.FileAppender;

import static ch.qos.logback.classic.Level.DEBUG;
import static ch.qos.logback.classic.Level.INFO;

import org.slf4j.LoggerFactory;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.core.rolling.RollingFileAppender;
import ch.qos.logback.core.rolling.TimeBasedRollingPolicy;
import ch.qos.logback.core.util.FileSize;
import net.logstash.logback.appender.LogstashTcpSocketAppender
import net.logstash.logback.encoder.LogstashEncoder

appender("STASH", LogstashTcpSocketAppender) {
  println "Setting [destination] property to 127.0.0.1:5000"
  destination =  "127.0.0.1:5000" 
  encoder(LogstashEncoder) {
   
  }
}

appender("ASYNC", AsyncAppender) {
  discardingThreshold=0;
  queueSize=500;
  neverBlock=true;
  appenderRef("STASH");
}

//root(DEBUG, ["ASYNC"])
root(INFO, ["ASYNC"])

logger("org.reactor.monitoring", DEBUG,["STASH"],false)

 3. Install ElasticSearch & Run

https://www.elastic.co/downloads/elasticsearch

elasticsearch-5.5.0/bin$ ./elasticsearch


 4. Install, Configure & Run Logstash


https://www.elastic.co/guide/en/logstash/current/installing-logstash.html

 Sample Logstash Configuration- logstash-filter.conf

input { 
  tcp {
   port => 5000
   codec => "json"
  }
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  date {
    match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

output {
  elasticsearch { hosts => ["localhost:9200"]
          index => "dlogs-%{+YYYY.MM.dd}"
          document_type => "log"
           
         }
  stdout { codec => rubydebug }
}

Run Logstash

logstash-5.5.1$ bin/logstash -f logstash-filter.conf


Sample Console Output

{
     "@timestamp" => 2017-07-30T15:28:37.792Z,
          "level" => "INFO",
           "port" => 52778,
    "thread_name" => "hz.ShutdownThread",
    "level_value" => 20000,
       "@version" => 1,
           "host" => "127.0.0.1",
    "logger_name" => "com.hazelcast.core.LifecycleService",
        "message" => "[10.180.35.234]:8701 [hibernate] [3.7.3] [10.180.35.234]:8701 is SHUTDOWN",
           "tags" => [
        [0] "_grokparsefailure"
    ]
}

 5. Insall & Run Kibana

https://www.elastic.co/guide/en/kibana/current/install.html

kibana-5.5.1-linux-x86_64$ ./bin/kibana 

6. Go to Kibana Dashboard

http://localhost:5601/app/kibana#/discover?_g=(refreshInterval:(display:Off,pause:!f,value:0),time:(from:now-15m,mode:quick,to:now))&_a=(columns:!(_source),index:'dlogs-*',interval:auto,query:(query_string:(analyze_wildcard:!t,query:'*')),sort:!('@timestamp',desc))

Master Data Management for Distributed Applications

Requirement Criteria

1. Self Load Balancing.

There are 100 products and those products need to be processed by each node without duplicating.

Example:


There are two nodes which need to process all the products as 50 products each at a time. So Node A will process 50 products while Node B will process remains 50 products.


2. High Availability and Fault Tolerant

If one node goes down remains should process all the products.

Example:

Let's say out of two nodes one nodes crashed so remain other node need to process all the 100 products until crashed node get recover.

3. Local (L1 cache) cache should be maintained by each node.

By doing this we can save unwanted round trips to Database for master data reading. Same time each node only need few Database connections to get master data from centralized DB. This is a micro services friendly approach when you use centralized DB for master data management.

4. Once master data got changed that should be reflected (synched) in each Local Cache.

Each node should process with latest data.

5. Distributed Map should maintain bare minimum information within its cache.

This will help to maintain performance and stability of distributed Map.


Architecture Explained

1. We used Hazelcast as a distributed in memory cache.

Using hazelcast distributed map we share all the products among the nodes.

2. Again we use Hazelcast for application clustering so each node will act as hazelcast cluster node. All the cluster communication and cluster management will done by hazelcast itself.

3. We used Hibernate as an ORM tool.

Hibernate Session act as a Local Cache so basically we just have to use it's API for L1 cache management.

4. In here Kafka act as data pipeline for the whole architecture.

So once master data got changed by Node.JS admin panel it will be sent to Kafka as a JSON.


Node which enabled DB write permission will update DB with necessary changes and same time update Distributed map with business key and primary key.


5. Each hazelcast cluster node acting as a consumer for Kafka under different consumer group.

Hazelcast Cluster == Kafka Consumer Group

So one of cluster member which disabled DB write permission will consume same JSON and updated relevant distributed map with master data record business key and primary key. By doing that rest of the members in the same consumer group (cluster) will get to know the changes and updated their L1 cache after reading changes from DB.