Fast forward transformation process in data science with Apache Spark

Data Curation :

Curation is a critical process in data science that helps to prepare data for feature extraction to run with machine learning algorithms. Curation generally involves extracting, organising, integrating data from different sources. Curation may be a difficult and time consuming process depending on the complexity and volume of the data involved.

Most of the time data won't be readily available for feature extraction process, data may be hidden is unobstructed and complex data sources and has to undergo multiple transformational process before feature extraction .

Also when the volume of data is huge this will be a huge time consuming process and can be a bottle neck for the whole machine learning pipeline.

General Tools used in Data Science :

R Language - Widely adopted in data science with lot of supporting libraries

Mat lab - Commercial tool with lot of builtin libraries for data science

Apache Spark - New, powerful and gaining traction, Spark on Hadoop provides distributed and Resilient architecture help to fasten the curation process by multiple times.

Recent Study

One of my project involved curing and extracting the features from huge volume of data in natural language conversation text. We started with using R programming language for the transformation process, R language is simple with lot of functionalities in statistics and data science space but has limitations in terms of computation and memory and in turn efficiency and speed. We tried to migrate the transformation process to Apache Spark and observed tremendous improvement in the performance of transformation, We were able to bring down the time for transformation from more day to almost an hour of time for huge volume of data.

Here are some of the benefits that I would like to highlight the benefits of Apache Spark over R.

Effective Utilization of resources:

By default R runs in a single core and is limited by the capabilities of the single core and memory usage. Even though you have multi core system R is limited with using only one core, for memory it has the process limitations of a 32 bit R execution with virtual memory user space of 3 GB and for 64 bit R execution limited to amount of RAM. R has some parallel lib packages that can help to span the processing to multi cores.

Spark can run in distributed form with the processing running on executors with each executor running on its own process utilizing the cpu and memory.Spark brings the concept of RDD (Resilient Distributed Dataset) to achieve distributed , resilient and scalable processing solution.

Optimized transformation:

Spark has the concept of Transformation and Actions where the transformation perform lazy evaluation of job execution until an Action task is being called and intern brings optimization when multiple transformations are involved before an Action task which leads to transferring the results back to the driver program

Integration to Hadoop Eco System

Spark integrates well in the Hadoop ecosystem with yarn architecture and can easily bind to HDFS , multiple NOSQL database like HBase, Cassandra etc.

Support for multiple languages:

Spark API's has support on multiple programming languages like Scala, Java and Python

Google Auto Awesome Video - Is it a machine learning solution ???

Wish you all a very happy and a wonderful new year 2016 !!!

Happy to start the year with a blog covering some aspects on machine learning and this post is actually an inspiration from the new year eve celebration.

In the new year eve celebration with friends and I captured some photo moments using google photos app. The next day morning when I woke up I got a notification in mobile , would you like to review and save the video made out of photos in the new year eve event with some nice background music added, in google terms they call as Auto Awesome videos in google photos app.

I am happy to see the video that has been made automatically and ready to share , there is also a manual mode where we can customize photos for the video. But my interest is on the automatic creation and started thinking how this design could have been ?

At first cut I was able to sense this could potentially be a machine learning implementation and with my limited data science knowledge I thought to do provide some guess work on how this could have been designed while running at a large scale for millions of tenants at the server end

Let us understand the requirement in detail , given a collection of images we have to perform the following

Categorize the images into groups and pick the group corresponding to a specific event say the new year celebration in this case.

To improve accuracy , check and eliminate any irrelevant images that went into the group by error

Judge the mood of the event and add appropriate background music

Now let us analyse the type of machine learning solutions that could have potentially been used for this design

First part of the problem is towards categorizing the images into groups based on some parameters , Clustering algorithm could be a best fit to perform this. Given a specific dataset clustering helps to categorize the dataset into different partitions based on features of data, In our scenario grouping could be based on time of photo taken but I have seen cases where grouping is done based on image background and persons involved.

Next is to eliminate outliers in the grouping ,some photos might have accidentally went in to the group. Algorithms like anomaly detection can be executed to eliminate those outlier images in the collection.

Final step is to understand the mood of the images and add relevant background music to the video, sentiment analysis algorithm on pictures could potentially help to understand the mood of the images.

Disclaimer : This is purely my own guess work of the design and google might have done in a different way :)

NOSQL with RDBMS fallback:

NOSQL adoption becoming prominent across different critical applications to reap the benefits of performance, fault tolerance, high availability for bigger volume database needs. While migrating to NOSQL one of the risk that architects feel is what if the application gets into some unseen issues and take more time to fix , as NOSQL adoption is not battle tested across different domain and sectors and how to design some fallback strategy.

Few factors that people may think while migrating to NOSQL

What will happen if we get into unexpected errors in production and if takes more time to fix?

What if the product vendors itself haven’t faced such scenarios?

What if I have reporting or other dependent system that are well integrated with RDBMS and difficult to migrate from RDBMS in the current phase?

Architects would like to design some fallback option as RDBMS where application can switch to RDBMS on unrecoverable NOSQL issues. This raises few questions in mind on how to design the same.

How do I sync up data both in NOSQL and RDBMS for a high data volume without losing the order of update?

How do I sync up without adding much overhead to application? Synchronous update to both NOSQL and RDBMS will be too much of overhead...

How to reliably update the data between the two systems without any loss?

What if RDBMS goes down and how can I design to sync up reliably even on failures?

I can think of design depicted below to address the same.

Few components involved in the design are Apache Kafka receiving the updates and Apache Storm process the data to update the same to RDBMS. Both of these system are designed to work for big data needs in a reliable and distributed form.

Apache Kafka is a high performance message queuing system. Application pose the messages (Insert / Update / Delete) to Kafka message queue. To improve the performance with parallel processing the queue can be partitioned by table / region / logical data design as per the NOSQL model.

Apache Storm is a real time processing engine that can consume message through Spout component, do some processing through bolts and update the data to RDBMS. Storm has topologies to process guaranteed data processing, transactional mode of commitments that makes it suitable to handle partial failures and during commitments.

Benefits:

Information can be updated in an asynchronous way to RDBMS reliably.

Apache Kafka providing high available, reliable, partitioned queuing system fits best to handle huge data volume.

Storm doing real time processing for Kafka messages on the partitioned queue and provides a reliable way to update RDBMS

On RDBMS failures Kafka will persist the messages and Storm can continue to sync up the messages when it is comes back.

Next Generation Enterprise Application Architecture

New generation applications are architectured not only with the goal of desiging the application functionally and performing stable but also focuses on different aspects that are becoming critical

Scalability – Elastic scalability for all the layers of the application including data tier

Fault Tolerance - Ability to handle failure smartly and avoid cascading failures from and to dependent systems

High Availability – Ability to have application highly available on all layers including database even on data center failures

Efficient utilization of Infrastructure - Ability to scale up and down on demand

Faster access to underlying data on high load and data volumes

Ability to handle different data formats efficiently

Few reasons that are tied to this evolution are the need and benefits towards cloud adoption ( could be either private or public cloud ) and the need to handle huge data volume with faster response on the data tiers

	Benefits	Solutions
Physical -> IaaS -> PaaS	Elastic Scalability High Availability Efficient Infrastructure utilization Zero downtime deployment	VMWare , Open Stack – Private Cloud IaaS AWS, Azure – Public Cloud IaaS , PaaS Cloud Foundry – PaaS on private and public cloud
Circuit Breaker	Fault Tolerance Better failure handling Avoid avalanche failures	Netflix Hystrix Polly
Service Registry	Registry for dynamic instance scaling	Netflix Eureka Apache Zookeeper
Intelligent Load balancing	Intelligent Load Balancing utilizing the elastic scaling and self-discovery	Netflix Ribbon F5 Nginix
Search	Quick search needs from huge data sets, full text search, pattern matching	Elastic Search Solr
Data Grid	Faster read write data, Reduce read / write overhead to database, high availability to data	Coherance Gemfire Membase
Queue	Reliable data transfer across different data layers	Kafka RabbitMQ JMS
NoSQL	Big data – database needs Heavy Read / Write on high data volumes Faster response needs on the data High Availability on data Fault Tolerance on data Distributed database Scalable database	Couchbase MongoDB HBase Cassandra Graph DB ( Titan, OrientDB )
Hadoop	Distributed file processing and storage ecosystem High speed batch (MapReduce) / real time ( Storm, Spark ) processing	Different Hadoop distributions like Hortonworks, CloudEra,MapR

Kafka messaging system

Apache Kafka :

Kafka is an open source message queuing solution under Apache project, Kafka is new when compared to existing queue solutions like RabbitMQ, ActiveMQ, AWS SQS on product maturity but is quickly gaining momentum due to its features. In this post we will analyze some features of Kafka to see why it is gaining attention in the market.

The demand for processing huge data sets is growing everyday across enterprise systems and data is being processed in batch or real time and the queuing systems play an important role in connecting the data from source system / producer to destination / consumers. With huge dataset in transit enterprise are looking for message solution that can provide high throughput per second , scale horizontally, provides high availability and integrate well with other solutions.

Scalability:

This is one of feature where Kafka gets edge over other solutions, the ability to scale horizontally, Kafka achieves it by means of partitioning. We can set the number of partition while defining a topic (queue) and these partitions will get distributed across the broker nodes in the cluster and hence when we want to scale the system we can add more broker nodes and hence the partitions get realigned across the added broker nodes.

Fault Tolerance and High Availability:

Kafka achieves high availability by means of replication the partitions get replicated across different broker nodes and Kafka uses Zookeeper for its co-ordination. When a broker node goes down zookeeper co-ordinates so that the data is continued to be served from the replicated broker node partition and hence high availability for data is achieved.

Unit of Order:

Kafka guarantees unit of order delivery at each partition level and messages posted across different partitions are not guaranteed to be in order.

Reliability & Guaranteed delivery:

Kafka provides reliability to the message delivery and has options of synchronous and asynchronous acknowledgements for the message delivery.

Integration with Big Data solutions:

Kafka comes as part of Hadoop distributions and integrates with Hadoop map reduce for bulk consumption in parallel, for real time stream processing needs Kafka has good integration with systems like Apache Storm and Spark.

Reference : Kafka

Build your own monitoring solution for couch base

Recently i was trying to build a monitoring solution for couch base , i followed a simple approach that worked out well, thought to share the same in this post.

Requirements for the solution

Simple solution that can collect metrics from the http stats endpoint of couch base

Script based solution that can customized by operation team

Visualization dashboards

No additional software installation on couch base servers

Solution Needs

An light weight app server that can collects metrics on regular interval

A persistence layer that stores the data

A visualization tool that can bind well with the persistent data

Solution Architecture

Solution Highlights

Solution collects json stats data which has thousands of metrics as a whole and stores it to elastic search

NodeJS is a light weight server based on java script

Couch base stat endpoint exposes JSON based metrics and elastic search works well with storing JSON data

Kibana provides nice visualization for elastic search through different charts

NodeJS provides built libraries to elastic search

Hosting NodeJS, elastic search, Kibana are very simple, you can setup easily all of these components in few minutes through dockers

Elastic search is highly scalable

The approach can be applied for any monitoring where metrics are exposed through json format

Please find the reference of the solution under github

Two Phase Commit

Bottlenecks in database layer

Database has been seen as a most common place of bottleneck for performance across different tiers of the application. Few possible reasons restricting RDBMS performance

RDBMS not able to scale horizontally

Locking at row level / data page level / table level during database transactions

NoSQL on the rescue

I have mentioned about NoSQL data stores in my previous blog which achieves horizontally scalability distribution . In this blog I would like cover how transactional behaviour is achieved with high performance in NoSQL

Transactions in RDBMS

Let us try to understand how transactions operates in RDBMS,transactions with ACID (Atomicity, Consistency, Isolation and Durability) compliance executes all the actions involved in the transaction in a single step, if all the actions succeeds it commits the changes otherwise all the changes are revoked. To achieve this locking happens across the tables and hence performance becomes bottleneck

Let us take a simple transaction in order placement and analyse , An simple order management transaction involves two tables involving order and billing.

Confirm the product for the order by decrementing a count in the product catalogue

Confirm billing for payment

If payment succeeds, transaction as a whole has to be committed . If payment fails for some reason , change in product catalogue has to be revoked to original state so that it is available for others to consume. RDBMS achieves this whole process as a single step by locking these tables until transaction is completed and hence gets the ability to commit or revoke at the end of transaction, but this gives a overhead in performance as these tables gets locked and any read/ write on those are kept on hold unless stale read is enabled

Restrictions with NoSQL

Let us understand restrictions in NoSQL towards achieving this type of transactions

NoSQL provides locking at row level and not across rows, tables etc

With adoption of polyghot persistence and distributed transactions we may need to perform a transaction across different datastores as well

Two Phase Commit

Two Phase Commit is an approach followed in NoSQL to achieve transaction like behaviour. As the name mentions transactions happens in two phases with the ability to commit or revoke the changes made in phase 1 during phase 2. The approach introduces a additional component transaction manager which helps to commit or roll back the changes made in each phase of the transaction

Advantages with Two Phase Commit approach

Provides high performance with transactions

Ability to retry for failure portions of the transactions ( interesting )

Provides distributed transaction like capabilities across data stores

My personal experience with Two Phase Commit

Recently I personally came to see a Two Phase Commit scenario handled in amazon for my order placement which became inspiration for this post.

I placed an order ( a laptop desk ) in Amazon where my order placement was received and I went for sleep.Looks the payment got failed for some reason. Next day morning I got a notification to retry my payment, in this case Amazon instead of revoking the order placed ,Amazon holded the order for additional time say for a day or two and provided option to retry payment failure.

My order confirmation

Payment retry for my order

I believe Amazon has implemented some form of two phase commit to achieve this, I was personally happy with the way amazon handled my payment failure as the order was not revoked and i was given retry option later to complete the order with my laptop desk was still reserved for me.

This also opens the door for other mode of payments like cash on delivery etc.

Few links on Two Phase Commit

Star Bucks Approach for Performance

MongoDB - how to perform 2PC