MapR Spark Certification tips

I recently cleared MapR spark certification and would like to share some tips as I was asked to do so, (here you go my friends)

I divided this blog into 3 sections. 

• prerequisite for exam
• exam topics and must cover material
• tips (don't ignore the topics at the end of this blog please)

Prerequisite 

First and foremost  -  Work on Spark and Scala for at least a year before attempting the exam. below points summarize the need.

• You should have basic knowledge of distributed functional programming
• Hands-on experience on spark 
• Have good exposure to Scala programming(not expecting to be expert but read the code and answer sensibly).

Exam topics and must cover Material

Lots of programming questions in the exam, code snippet is provided and ask solve it and answer. If I remember correctly only 10% questions were theoretical ( like true/false or which algorithm to use kind).


I referred lot of materials(online books/videos/ edx courses in last 2 years) for my preparation but if I want to zero-in for what should be the mandatory for MapR certification - here is the list you should not miss any bit and I suggest to go over 4-5 times before taking the exam. 

• Instructor and Virtual Instructor-led Training(Training ppt and Lab guide)
• DEV 360 – Developing Spark Applications
• DEV 361 - Build and Monitor Apache Spark Applications
• DEV 362 - Spark Streaming, Spark MLLib - Machine Learning, Graphx
• Book - Learning Spark
• Spark official documentation
• pay more attention to RDD, Closure, Accumulator, Broadcast variables.  
• http://spark.apache.org/docs/latest/quick-start.html
• http://spark.apache.org/docs/latest/rdd-programming-guide.html
• MlLib - http://spark.apache.org/docs/latest/ml-guide.html

Topics covered in the exam 

Topic Name		Your Score
Load and Inspect Data in Apache Spark		xx%
Advanced Spark Programming and Spark Machine Learning MLLib		xxx%
Monitoring Spark Applications		xx%
Work with Pair RDD		xx.x%
Spark Streaming		xx%
Work with DataFrames		xx%
Build an Apache Spark Application		xxx%

Tips  - 

Normally when anyone start preparing for the exam - the good start will be to go through below link 

https://mapr.com/blog/how-get-started-using-apache-spark-graphx-scala/assets/spark-certification-study-guide.pdf

The question on this guide is way to basic comparing to the real exam. the exam was much-much harder.

• Lots of question on core concepts of RDD and pair RDD
• Dataframes are the next important
• About 25% questions on Spark Streaming and Spark MLLib so prepare well on 

You don't want to ignore any of the below topics at any cost

Silent topics which you don't want to get as surprise in exam

• Accumulator and Broadcast variables
• Scala Closures
• Narrow and Wide Dependencies
• Partitioning  
• Formating questions – saveAsTextFile() – need to save without bracket/parenthesis
• Prepare well for mkString(“,”) and formating 
• flatMap functions
• MapPartitions
• There was a question on byKey transformation and also on hadoop streaming which I am not sure about. 

Hope this blog will help in your preparation. Please let me know or email me if you have any other questions. Happy Studying

At the end  - Here is my certification

Blockchain basics

 Blockchain is a continuously growing list of records which are linked and secured using cryptography.For use as a distributed ledger, a blockchain is typically managed by a peer-to-peer network collectively adhering to a protocol for validating new blocks. Once recorded, the data in any given block cannot be altered retroactively without the alteration of all subsequent blocks, which requires collusion of the network majority.

Why Block chain

Companies/parties today keep track of records of all transactions between all the parties that the business interacts and update as and when needed. this process is inefficient because 

• duplication of information with all the parties to update the ledger
• less transparent 
• less trusted transactions as data owned by one party and cant be guaranteed to be true.
• error-prone

The solution is to use distributed, secure, transparent and shared ledger - Block chain

• A shared ledger technology
• transparent transaction as all parties are involved/informed
• immutable chain - only appended

point to note  - Blockchain is still an emerging technology. Business owners need to start small and then look for more ways to grow and expand the use of blockchain networks.

How block chain applied to business network

Assets are classified into tangible(land, properties), intangible(Cash, loan). 

• The transaction records of assets are kept in the form of distributed ledgers(block chain).
• Flow of assets/transaction is governed by contract.

This assets can be tracked using this distributed ledger for transparency and maintain only single copy which is shared/endorsed by all party involved hence trusted.

You can see the life cycle of an asset

Blockchain in business 

Blockchain for business provide  - secure, shared ledger which one single record which is accessible for all party involved hence transparent. Business networks prioritize identity over anonymity. Assets are more diverse and important in a business network. A business network gets to choose who validates a transaction.

• All the member of business network share the common ledger on block chain. 

• Ledgers are replicated. 

• all member involved can view the transaction but only authorized members can update the transaction. 

The requirements for a blockchain for business are a shared ledger, smart contract, privacy, and trust.

1. shared ledgers

2. privacy services(who can see what and update the information)

3. Trust - transaction are endorsed by relevant participants 

4.    4. Contract - common/shared business process.

        For example, for financial services network, a business network that runs on a blockchain can speed up transaction processes and audits. That in turn reduces costs and can lead to greater customer satisfaction. A business that runs a supply chain network can benefit from blockchain by reducing errors in shipments, have better tracking or materials, and reduce the risk of illicit tampering of records.

Blockchain for business has several advantages:

• Saves time
• Removes cost
• Reduces risk
• Increases trust

Use cases of block chain

1. Reference data

2. Supply Chain

3. Trades(Diamond life cycle)

blockchain and bitcoin

Bitcoin is an unregulated shadow-currency and was the first popular blockchain application. The Bitcoin application works in an anonymous network, so no one knows who the participants are.

Bitcoin blockchain is protected by the massive group mining effort. It's unlikely that any private blockchain will try to protect records using gigawatts of computing power — it's time consuming and expensive

Spark Mlib Basics

MLlib Machine Learning Library

Spark MLlib is a distributed machine learning framework on top of Spark Core that, due in large part of the distributed memory-based Spark architecture, is as much as nine times as fast as the disk-based implementation used by Apache Mahout (according to benchmarks done by the MLlib developers against the Alternating Least Squares (ALS) implementations, and before Mahout itself gained a Spark interface. Many common machine learning and statistical algorithms have been implemented and are shipped with MLlib which simplifies large scale machine learning pipelines, including:

1. summary statistics, correlations, stratified sampling, hypothesis testing, random data generation
2. classification and regression: state vector machines, logistic regression, linear regression,      decision trees,naive Bayes classification
3. collaborative filtering techniques including alternating least squares (ALS)
4. cluster analysis methods including k-means, and Latent Dirichlet Allocation (LDA)

Machine Learning(definition) - Constructing and studying methods that learn from and make predictions on data. 

Terminologies
  Observations - (data points) item or entities used for learning or evaluation. 
  Features - attribute used to represent an observation. 
  Labels -  value assigned to observation
  Training and Test data - observation used to train or evaluate a learning algorithm. 

so if consider observation as email than feature would be date, importance, key words in subject or body of an email and Labels would be spam or not-spam. Test and learning data would be set of emails.

Supervised learning - learning from labeled observation examples classification and regression
Unsupervised learning - learning from unlabeled observation examples clustering and dimensionality reduction. 

Flow -raw data ->feature extraction-> supervised learning ->evaluation-(satisfied)-> prediction

MLlib consists of two packages.

• spark.mllib  
• spark.ml

When using pyspark, you'll find them in the pyspark.mllib and pyspark.ml packages respectively
Spark.ml is a newer package and works with data frames. The algorithm coverage is similar between the two packages, although spark.ml contains more tools for feature extraction and transformation. ML Package contains two types of classes transformer and estimator.

Transformer is a class which takes dataframe as input and transform it to another dataframe. 

A transformer implement transform() function which is called on input dataframe. 

Examples :

• Hashing Term Frequency - which calculates how often words occur. It does this after hashing the words to reduce the number of features that need to be tracked.
• LogisticRegressionModel - The model that results from trying logistic regression on a data set, this model can be used to transform features into predictions.
• Binarizer - which changes a numeric feature into 1 or 0 given a threshold value.

An Estimator is a class that can take a DataFrame as input and returns a Tranformer.

It does this by calling it's fit method on the input DataFrame.

Note that Estimators need to use the data in the input DataFrame to build a model that can then be used to transform that DataFrame or another DataFrame.

Examples :

• LogisticRegression processes the DataFrame to determine the weights for the resulting logistic regression model.
• StandardScaler needs to calculate the standard deviations, and possibly, means of a column of vectors so that it can create a standard scalar model. That model can then be used to transform a DataFrame by subtracting the means and dividing by the standard deviations.
• PipeLine Calling fit on a pipeline produces a Pipeline model. The pipeline model only contains transformers. There are no estimators.

ML Pipeline  - Its a estimator that consist of one or more stages representing a reusable workflow.  Pipeline stages can be transformers, estimators or another pipeline.

             

transformer1->transformer2->estimator1

-----------------------pipeline-----------------------------

Loss functions define how to penalize incorrect predictions.

The logistic function asymptotically approaches 0 as the input approaches negative infinity and 1 as the input approaches positive infinity. Since the results are bounded by 0 and 1, it can be directly interpreted as a probability.

Feature engineering is the important part and we will discuss that next until enjoy learning. 

Spark Dataframes

Concept of dataframes  -  A data frame is a table, or two-dimensional array-like structure, in which each column contains measurements on one variable, and each row contains one case.

So, a DataFrame has additional metadata due to its tabular format, which allows Spark to run certain optimizations on the finalized query. 

                                                       or

A DataFrame is equivalent to a table in RDBMS and can also be manipulated in similar ways to the "native" distributed collections in RDDs. Unlike RDDs , Dataframes keep track of the schema and support various relational operations that lead to more optimized execution. 

Each DatafRame object represents a logical plan but because of their "lazy" nature no execution occurs until the user calls a specific "output operation".

                                                       

Dataframes are distributed across worker. It is the primary abstraction in Spark SQL and immutable. 

You can think of Dataframes as special type of RDD.  Dataframes keep track of the schema and support various relational operations that lead to more optimized execution.   An RDD, on the other hand, is merely a Resilient Distributed Dataset that is more of a blackbox of data that cannot be optimized as the operations that can be performed against it are not as constrained.

However, you can go from a DataFrame to an RDD via its rdd method, and you can go from an RDD to a DataFrame (if the RDD is in a tabular format) via the toDF method

In general it is recommended to use a DataFrame where possible due to the built in query optimization.

You can easliy switch from dataframes to RDD (1) or RDD to dataframes(2)

(1) val sampleRDD = sqlContext.jsonFile("hdfs://localhost:9000/jsondata.json")

(2)val sample_DF = sampleRDD.toDF()

The main disadvantage to RDDs is that they don’t perform particularly well whenever Spark needs to distribute the data within the cluster, or write the data to disk, it does so using Java serialization by default (although it is possible to use Kryo as a faster alternative in most cases). The overhead of serializing individual Java and Scala objects is expensive and requires sending both data and structure between nodes (each serialized object contains the class structure as well as the values). 

There is also the overhead of garbage collection that results from creating and destroying individual objects.

Spark 1.3 introduced a new DataFrame API which seeks to improve the performance and scalability of Spark. The DataFrame API introduces the concept of a schema to describe the data, allowing Spark to manage the schema and only pass data between nodes, in a much more efficient way than using Java serialization. 

There are also advantages when performing computations in a single process as Spark can serialize the data into off-heap storage in a binary format and then perform many transformations directly on this off-heap memory, avoiding the garbage-collection costs associated with constructing individual objects for each row in the data set. Because Spark understands the schema, there is no need to use Java serialization to encode the data

There are key data management concepts that you need to know.

• The first is that a data model is a collection of concepts for describing data.

• The second is that a schema is a description of a particular collection of data that uses a given data model

Relational database model is widely used model wherein Relation is a table with rows and columns and each relation has a schema defining fields in columns.

Instance - actual data at given time

#ofRows - Cardinality
#ofColumns - degree

• Each row is a dataframe is a Row object.

row = Row(name="Pawan", age=30)

and can be accessed via

row.name or row['name']

• Creating a dataframe from Python list

1. You create a data frame from a data source, from disk or from a Python object.

data = [('pawan',30),('sapan',26)]

df = sqlContext.createDataFrame(data)

 or assign the name of the columns(schema)

df = sqlContext.createDataFrame(data, ['name','age'])

2. You apply transformations to that data frame, like select and filter.

df2 = df.select(df.name, df.age)

or

df2 = df.filter(df.age > 28)

3. Cache some dataframe for reuse 

df.cache() //save it in memory and reuse it else it will read from the disk every time

4.  then you apply actions to the data frame,like show and count.

df2.collect(); //never use collect in production application - it will crash the driver machine(Out of memory) if it cant hold the data. Use show() or take() instead.

Point to note 

• When you're dealing with group data, count is a transformation, but when you're dealing with data frames, count is an action.
• Transformers runs at executors and Actions runs at executors and driver. 
• use UnionAll() function to combine the dataframes. 
• Cache if dataframes are reused.
• never use collect in production

When you use DataFrames or Spark SQL, you are building up a query plan. Each transformation you apply to a DataFrame adds some information to the query plan. When you finally call an action, which triggers execution of your Spark job, several things happen:

1. Spark's Catalyst optimizer analyzes the query plan (called an unoptimized logical query plan) and attempts to optimize it. Optimizations include (but aren't limited to) rearranging and combining filter() operations for efficiency, converting Decimal operations to more efficient long integer operations, and pushing some operations down into the data source (e.g., a filter() operation might be translated to a SQL WHERE clause, if the data source is a traditional SQL RDBMS). The result of this optimization phase is an optimized logical plan.
2. Once Catalyst has an optimized logical plan, it then constructs multiple physical plans from it. Specifically, it implements the query in terms of lower level Spark RDD operations.
3. Catalyst chooses which physical plan to use via cost optimization. That is, it determines which physical plan is the most efficient (or least expensive), and uses that one.
4. Finally, once the physical RDD execution plan is established, Spark actually executes the job.

You can examine the query plan using the explain() function on a DataFrame. By default, explain() only shows you the final physical plan; however, if you pass it an argument of True, it will show you all phases.

MemSQL - database for real-time analytics

MemSQL

MemSQL is relational, distributed, in memory SQL database platform for real-time analytics.

MemSQL is a proprietary closed-source distributed relational/sql database with both free community and paid enterprise edition

MemSQL enables sub-second refresh of dashboards with real-time data for improved accuracy of positions and risks to drive portfolio growth and customer satisfaction.


MemSQL is a two-tiered architecture consisting of aggregators and leafs.


Aggregators are cluster-aware query routers that act as a gateway into 

the distributed system.The only data they store is cluster metadata. 
The leaf nodes store and compute data .

Overview

• Distributed SQL database, fully ACID, JSON and Geo Spatial support
• Rely on main memory for storage.  It can spill to disk or pin data in-memory.
• Need lot of memory - its a trade off but its cheaper today (every year 40% decrease). Cache is a new RAM, RAM is a disk and disk is
  the new tape. Leverage SSDs (no random write). NVRAM or non-volatile RAM
  is coming sometimes soon.
• Use MySQL wire protocol.
• seemless integration with Kafka
• 10 million upserts per second

Architecture

• memsqld,
• aggregators
• leaf(holds partitions) nodes

Aggregator node stores metadata and leaf node data in partitions 

A partition is an indivisible slice of data that can be replicated and

moved around the cluster.

Agg and Leaf workload

Communication and Ports

Sharding -

insert into tab01 values ('geocode','profileid, .....)

hash("geocode|profileid") % partition  = partition no. (say 8)

Code Example - Spark Aggregates RDBMS and Memsql data

Road Map

RoadMap

MemSQL not for 

• MemSQL is not designed to be a persistence store (Hbase) or “data lake”
• MemSQL supports extremely fast, distributed “READ-COMMITTED” transactions,                 but it is not suitable for applications which require “SERIALIZABLE” transactions.

Dependency

<dependency> 

<artifactId>memsql-connector_2.11</artifactId>

<version>2.0.2</version> 

</dependency>

/*using memsql connector with Apache spark.  In a program,I tried to mix a streaming data coming to memsql and existing database table and join them and put the result for user dashboard. */

package com.valassis.msg.consumer;

import java.io.IOException;
import java.util.Properties;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
public class MemsqlRDBMSConsumer {

public static void main(String[] args) throws
InterruptedException,

IOException {

SparkSession spark = SparkSession.builder()

.appName("Java Spark MemSQL")

.config("spark.memsql.host",
"localhost")

.config("spark.memsql.port",
"3306")

.config("spark.memsql.user",
"root")

.config("spark.memsql.password",
"memsql123")

.config("spark.memsql.defaultDatabase",
"impower_pawan")

.getOrCreate();

/*

* Dataset<Row> views = spark.read()

*
.format("com.memsql.spark.connector") .option("query",

* "select user, HOST from
information_schema.users") .load();

* views.show();

* 

* views.printSchema();

*/

Dataset<Row> result = spark

.read()

.format("com.memsql.spark.connector")

.option("query",

"select * from
impower_pawan.geofootprintmaster")

.load();

result.show();

result.printSchema();

result.createOrReplaceTempView("memsql");

String url =
"jdbc:oracle:thin:@bruno_bruno1.val.vlss.local:1587:BRUNO1";

String table = "SDR_PM.PA_STORES";

Properties connectionProperties = new
Properties();

connectionProperties.put("driver",
"oracle.jdbc.driver.OracleDriver");

connectionProperties.put("user",
"sdr_pm");

connectionProperties.put("password",
"pm007");

Dataset<Row> jdbcDF2 =
spark.read().jdbc(url, table,

connectionProperties);

jdbcDF2.show();

jdbcDF2.createOrReplaceTempView("geo");

Dataset<Row> sqlDF = spark

.sql("SELECT b.id, a.geo_profile_id,
a.name, b.value FROM geo a, memsql b where b.id = a.geo_profile_id");

sqlDF.show();
}
}