Hands-On Machine Learning Recommender Systems with Apache Spark

Build a real Artificial Intelligence solution with real data

Change the world with Machine Learning

Nesto.TV and ConsultantsNetwork.com

Ernesto Lee, MS

http://www.Nesto.TV

http://www.LearningVoyage.com

Fort Lauderdale, Florida

Panama City, Panama

Hands-On Machine Learning Recommender Systems with Apache Spark

Build a real Artificial Intelligence solution with real data

Copyright © 2020 Consultants Network

All rights reserved. No part of this publication may be reproduced, distributed, or transmitted in any form or by any means, including photocopying, recording, or other electronic or mechanical methods, without the prior written permission of the publisher, except in the case of brief quotations embodied in critical reviews and certain other noncommercial uses permitted by copyright law. For permission requests, write to the publisher, addressed “Attention: Permissions Coordinator,” at the address below.

Nesto.TV

1918 Harrison Street, Suite 215

Hollywood, FL 33020

www.Nesto.TV

Ordering Information:

Quantity sales. Special discounts are available on quantity purchases by corporations, associations, and others. For details, contact the publisher at the address above.

Orders by U.S. trade bookstores and wholesalers.

Printed in the United States of America

[TITLE] : [SUBTITLE] / Ernesto Lee, Nesto.TV

ISBN [ISBN NUMBER HERE]

1. The main category of the book —Software—. 2. Subject category —Programming—

First Edition

http://www.learningvoyage.com

http://www.consultantsnetwork.com

http://www.nesto.tv

Authors

Ernesto Lee

Uzair Syed

Reviewers

Eric Johnson, Addison Jones, Larry Watkins

Project Team Leader

Ernesto Lee

Technical Editor

Ernesto Lee

Editorial Team Leader

Ernesto Lee

FORWARD

Those with the ability to solve problems and think from a solution-oriented approach will always be able to thrive and grow in our industry. While the products that we work on over the years will inevitably become stale and overcome by newer technologies, the drive to be better today than we were yesterday will always keep us moving in the right direction.

I am sure that you will find this book to be more reference than theoretical. As a result, it is intended to be used as guide that shows you exactly HOW to perform tasks while at the same time providing context. I hope you find this book to be useful.

Ernesto Lee

WHO WE ARE

Nesto.TV and Learning Voyage

Ernesto Lee:   Holds a Master’s Degree in Software Systems Engineering from Virginia Tech and a Bachelor’s Degree in Physics from Old Dominion University. He is presently completing his Doctorate Degree from Nova Southeastern University. Ernesto is responsible for working with organizations to enable them to realize the full business benefits of artificial intelligence and big data in solving complex business problems

Ernesto has been involved in several large–scale projects; he has consulted for several large companies in different domains like Healthcare, Banking, Manufacturing, and Retail.

Ernesto and his team have extensive experience and expertise in implementing business solutions for customers that leverage the right technology. Aside from being a full time student, he is the founder of LearningVoyage.com and Nesto.TV were he focuses on EdTech solutions in Machine Learning, Blockchain, Microservices, and Information Security.

Acknowledgement

Shirley L. Jones, Tyrone V. Lee, Devita Vanae Evans … gone but never forgotten

Write for Us

Nesto.TV and Learning Voyage continue to look for authors with both technical expertise and the ability to explain. We are currently looking for authors in the Machine Learning, Blockchain, Microservices, and Cybersecurity space but we are open to entertaining products in interesting verticals.

We are interested in working with you if you are an expert in your field first and foremost and you have the ability to deliver quality, original work. We definitely work with you to make sure that your project is as successful as it can be but that all starts with you contacting us at:

support@LearningVoyage.com

Feel free to write with queries but to maximize our interaction, please provide:

- Contact information

- A table of contents (of course)

- Resume

- Why is it that you are qualified to write this book

- Why would this book sell in the market

- A writing sample

Table of Contents

CHAPTER 1: INTRODUCTION TO BIG DATA & AI

Theory

This chapter is intended to provide a comprehensive introduction to recommender systems using Apache Spark / Machine Learning. Before we begin with recommender systems using Apache Spark, let’s have a brief overview of Big Data. To better understand Spark, we should know a little bit of history before the advent of Spark. We shall be looking at a quick introduction to Hadoop and MapReduce before we look at Spark.

An Overview of Big Data

Quick Introduction to Hadoop

Apache Hadoop is an open source distributed framework that allows storage and processing of large data (Big Data) sets across a cluster of commodity machines. Hadoop overcomes the traditional limitations of storing and computing of data by distributing the data over cluster of commodity machines making it scalable and cost-effective.

The idea of Hadoop was originated when Google released a white paper about the Google File System (GFS) - a computing model built by Google which was designed to provide efficient, reliable access to data using large clusters of commodity hardware. The model was then adopted by Doug Cutting and Mike Cafarella for their search engine called “Nutch”. Hadoop was then developed to support distribution for the Nutch search engine project by Doug Cutting and Mike Cafarella. It is often asked, “what does the name Hadoop mean”? There is no significance for the name and it is not an acronym either. Hadoop is the name that Doug Cutting’s son gave to his yellow stuffed elephant. The name is very unique, and easy to remember. Not only does the name Hadoop have no real significance but also its sub-projects tend to have such names which are based on names of animals like Pig for the same reasons. They are unique, not used anywhere else and are easy to remember.

Why Hadoop?

Companies today have been realizing that there is lot of information in unstructured documents spread across the network. A lot of data is available in the form of spreadsheets, text files, e-mails, logs, PDF’s and other data formats that contain valuable information which can help discover new trends, designing new products, improving existing products, knowing customers better and many other reasons. Data is increasing at a staggering rate, beyond limits never before seen and there are no signs of slowing down. To deal with such data, we need a reliable and low-cost tool to meaningfully process it. Therefore, we use Hadoop. Hadoop helps us process all this Big Data which is present in a variety of formats reliably, faster, with more flexible and in a cost-effective way.

Let us see why Hadoop is so popular and what it has in store for you.

●  Scalable: Hadoop is scalable, meaning; you can just start from a single node server and eventually increase more nodes as you need more storage and more computing power.

●  Fault-Tolerant: Hadoop helps prevent the loss of data. All the data which is stored in Hadoop Distributed File System is broken into blocks and stored with a default replication factor of 3. While processing data, if a node goes offline, the process continues as the data still exists in other nodes.

●  Flexible: Hadoop does not require a schema. Hadoop can process unstructured, semi-structured and structured data from any kind of source or even from multiple sources.

●  Cost effective: Hadoop does not require expensive high-end computing hardware. Hadoop works well with a cluster of commodity machines by parallel computing.

Quick Introduction to Hadoop Distributed File System

Hadoop Distributed File System (HDFS) is a File System which extends over a cluster of commodity machines rather than a single high-end machine. HDFS is a distributed, large-scale storage component and is highly scalable. HDFS can accept node failures without losing data. HDFS is widely known for its reliability. Let us now check out why HDFS stands out in the crowd when it comes to Distributed file systems.

Reliable Data Storage

HDFS is very reliable when it comes to data storage. Whatever the data stored in HDFS is replicated, it is, by a default replicated by a factor of 3. That means, even if a machine fails, the data will be still available in two other machines.

Cost Effective

HDFS can be deployed on cluster of commodity hardware and can drastically decrease hardware costs. High end expensive hardware is not required by HDFS to function.

Big Datasets

HDFS is capable of storing Petabytes of data over a cluster of machines where a file can range from Gigabytes to Terabytes of size. HDFS is not designed to store huge number of small sized files as the file system meta data is stored in memory on the NameNode.

Streaming Data Access

HDFS provides streaming access to data. HDFS is best suited for batch processing of data and is not suitable for interactive processing. HDFS is not designed for applications which require low latency access to data such as Online Transaction Processing (OLTP).

Simple Coherency Model

HDFS is designed to “ write once and read many times” access model for files. Appending the content to files is supported at the end but cannot by updated at arbitrary points and it is also not possible to have multiple writers. Files can only be written by a single writer.

Block Placement in HDFS

Hadoop is designed in such a way that the first block replica is placed on the same node as the client and the second replica is placed on a different rack to that of first replica. The third replica is placed on a random node on the same rack as of second replica. If the replication factor is more, random nodes in the cluster are selected to be placed on the replicas. If a client running outside of the cluster stores a file, a random node (that isn’t busy) is picked to place the first replica. This way, if a node fails, the data is still available on the other nodes of the cluster. Said differently, if a rack fails, the data is still intact.

HDFS Architecture

HDFS is a Master and Slave architecture, in which the Master node controls and assigns jobs to all its slave nodes. The following terminologies are used to describe the Master and Slave nodes.

The Master Nodes in HDFS are:

●  NameNode

●  Secondary NameNode

The Slave Nodes in HDFS are:

●  Data Nodes

These nodes are the core serving roles in HDFS architecture. Let us now look in detail about the roles of each Node and understand them better.

NameNode

The NameNode is an HDFS daemon which controls all the Data Nodes and handles all the File System operations such as creating a directory, creating a file or reading and writing a file. The NameNode is responsible for managing the File System namespace image. It holds the image in memory, representing how the File System looks. It also maintains the meta data for all the blocks of files in the File System and also tracks the replication value, so it knows the locations of blocks stored on Data Nodes within the cluster. The meta data is not stored on the disk and is recreated each time when it starts. The NameNode stores all this information persistently on the local disk in the form of a namespace image and edit log. The NameNode is the  single point of failure in the Hadoop cluster. If the NameNode fails, the entire cluster fails.

Data Nodes

Data Nodes are the slave machines controlled by the NameNode, that actually does all the block operations. Data Nodes store and retrieve blocks when asked to do so by the NameNode and periodically informs the NameNode with the lists of blocks they store by sending heartbeats. Data Nodes replicate the data physically when instructed to do so by the NameNode on where and how to replicate.

Secondary NameNode

Secondary NameNode, as its name implies, is not exactly the Secondary NameNode. The secondary NameNode is not a high availability solution and does not automatically takeover the responsibilities of NameNode on failure. Its main role is to create checkpoints and take the backup of the NameNode periodically. It is like a backup solution to the NameNode. The hardware specifications of secondary NameNode should be similar to that of NameNode. In case of a NameNode’s failure, the secondary NameNode can be manually configured to work as a primary NameNode. This is not a high availability solution.

Introduction to MapReduce

MapReduce is a programming model for processing large amounts of data stored in distributed file systems such as HDFS. MapReduce is about the low-level programming and therefor programs are written in the corresponding low level languages such as Java, Python, Scala, Ruby and others.

Architecture of MapReduce

MapReduce, similar to HDFS is a master-slave architecture but instead of the NameNode in HDFS, we have a JobTracker. Instead of Data Nodes in HDFS, we have Task Trackers in MapReduce. Unlike NameNode and Data Nodes, the JobTracker and Task Trackers are not physical hardware components but Java programs running on their own JVM’s inside the machines.

The detail description of MapReduce daemons is as follows:

JobTracker

The JobTracker is the MapReduce daemon which is the master to the Task Trackers. The role of JobTracker is to receive job requests from Hadoop clients and assign work to task trackers on Data Nodes. The JobTracker queries the location of the data on the Data Nodes and tries to assign task trackers on Data Nodes where the data is present locally, so as to achieve data locality. If the node where the data is present locally is busy, the Job Tracker assigns tasks to the task tracker within the same rack. If the task tracker fails for any reason, the JobTracker will assign the same task on another task tracker as the data is replicated across the cluster on other nodes as well.

The JobTracker is the single point of failure, similar to that of the NameNode. If the JobTracker fails, all the task trackers fail and there will be no tasks running. So, it is wiser to ensure there are ample resources for the machine which runs the JobTracker so as to decrease the chances of failure.

Task Trackers

The Task Trackers are slaves to JobTracker which do the actual work. TaskTrackers accept tasks from JobTracker and performs the tasks. The task trackers send the status or progress of the tasks to the JobTracker in the form of heartbeats so that the JobTracker can know that the task trackers are performing as they should and they have not failed. The task trackers also send heartbeat messages to the JobTracker about the free slots available for processing of tasks.

The failure of a task tracker is not as serious as that of the JobTracker as the JobTracker can always assign the failed task to another task tracker.

Processing Data with MapReduce

MapReduce consists of two major phases through which the data is processed. The two major MapReduce phases are:

●  Map phase

●  Reduce Phase

Both the phases have key-value pairs as input and output. The data types for the key-value pairs can be chosen by the developer and the developer has to specify a map function and a reducer function, which is the logic for processing the data in MapReduce. The map function provides the logic for the mapping tasks and the reduce function provides the logic for the reduce task.

Let us now look at the flow of data and the various stages of data processing in MapReduce.

●  The input to map task is in the form of a split. A split is a fixed chunk of data based on the inputFormat and should not be confused with HDFS block. Blocks belong to HDFS and splits belong to MapReduce. A block is the smallest size of data stored in HDFS whereas a split is an input to a map task. The optimal split size will be equal to the block size. A map task can process one input split at a time.

●  A map task processes the splits using the map function in parallel and produces an output. The output from the mapping tasks is not stored in HDFS but rather on the local disk because the map output is just the intermediate result and there is no need to save it on HDFS with replications. The map output is only saved in local disk until the reducer has produced the final result. If we have specified no reducers for the job, the map output will be the ultimate result and will be stored in HDFS with replications.

●  The output from all the map tasks are merged, sorted and partitioned. Merging is the process where the data from all the mapping tasks are merged together. Sorting is the process where the mapping task output is sorted based on the key. Partitioning is the process where the data is divided based on the keys so that the values from all the keys should go to one reducer. Partitioning is useful when there are multiple reducers used. The map task output will then be fed to the reducer as input.

●  The reducer then processes the map output using the reducer function and produces the desired result. Unfortunately, reducers cannot take the advantage of data locality and will be fetched across the network. The number of reducers is not set automatically. The number of mappers on the other hand is automatically set based on number of splits. The output produced by reducers is the end result and will be stored in HDFS with replications.

Let us consider an example with a few words in a document to understand the concept of MapReduce in a pictorial representation.

Consider a file which consists of search terms for a website with each search term in a separate line as shown below.

Input file

Ernesto Lee

Ernesto Lee Website

Learning Voyage

Learning Voyage Website

Ernesto Lee Blog

Let us assume that each line is an input to different map tasks. Please note that, the above assumption is just for understanding the concept and in reality, each map task processes much more data. The input file is broken into splits so as to feed to the map task. The file is split line by line as each line corresponds to a search term which will be sent to the mappers. The splits of the file are as shown below:

Input Splits

Ernesto Lee

Ernesto Lee Website

Learning Voyage

Learning Voyage Website

Ernesto Lee Blog

These input splits are further broken down into individual words and then submitted to the mapper in the form of key value pairs. The key will be the word and value will be 1 as shown below.

Ernesto            1

Lee            1

Mapper

Ernesto            1

Lee            1

Website      1

Learning      1

Voyage            1

Learning      1

Voyage            1

Website      1

Ernesto            1

Lee            1

Blog            1

At this point, all the intermediate data from map tasks output goes through the shuffle and sort phase. All the relevant words are brought together in our example as shown below.

Website            1

Website            1

Shuffle & Sort

Learning            1

Learning            1

Voyage            1

Voyage            1

Ernesto            1

Ernesto            1

Ernesto            1

Lee            1

Lee            1

Lee            1

Blog            1

When the intermediate data is processed through the shuffle and sort phase, it is time for the reducer to produce the end result using the logic in the reducer function. After the shuffle and sort phase, the partitioner makes sure that the key value pairs with the same keys go to the same reducer if there are multiple reducers set.

The reducer will aggregate the input and produce the following output.

Ernesto            3

Reducer

Lee            3

Website      2

Voyage            2

Learning      2

Blog            1

The output from the reducers is concatenated to have a single file which contains the end result. The end result is as shown below.

Ernesto            3

Lee            3

Learning      2

Blog            1

Voyage            2

Website      2

Result

The result shows the occurrences of each word in a given file.

Finally let us conclude the quick introduction on HDFS by looking at the 3V’s of Hadoop which also serves to summarize the Hadoop capabilities.

3V’s of Hadoop

Hadoop can be better described with 3V’s. The 3V’s of Hadoop are as follows

●  Volume: Hadoop is designed to process large amounts of data (ranging from hundreds of Gigabytes to Petabytes). There are lots of Petabyte datasets available today and Exabyte datasets are starting to occur. Hadoop excels while processing large amounts of data rather than small data.

●  Velocity: Hadoop is designed to ingest data at higher speeds from multiple sources. Hadoop uses the distributed framework for parallel processing which in turn decreases the time taken to complete a job. Hadoop brings computation to data rather than bringing data to computation which bottle necks the network.

●  Variety: Hadoop can process data in structured, semi-structured and unstructured forms. There are no restrictions on schema types. Unlike relational database management systems, Hadoop has the “schema on read” capability. There is no schema required while writing to HDFS. The schema can be parsed at read time.

Traditional computing models lack these features but this is where Hadoop excels by providing you with more power to explore your data.

Now that we have had a quick introduction to Hadoop, let us shift our focus to the main topic of our discussion, Apache Spark.

Introduction to Spark

What is Spark?

Apache Spark is an open source, fast and unified parallel large-scale data processing engine. It provides a framework for programming through distributed processing of large datasets at high speed. Spark supports most of the popular programming languages such as Java, Python, Scala and R. Spark is scalable, meaning, it can run on a single desktop machine or a laptop to a cluster of thousands of servers. Spark provides a set of built in libraries which can be accessed to perform data analysis over a large dataset. However, if your requirements exceed the capabilities present in the built in libraries, you can write one or explore countless external libraries from the myriad open source communities on the internet.

Why Spark?

Why use Spark when we have Hadoop? Well, Spark excels as a unified platform for processing huge data at very high speeds for various data processing requirements. Spark is an in-memory processing framework. Spark is arguably mentioned as the successor of Apache Hadoop. Let us briefly discuss the advantages of Spark over Hadoop.

With the Hadoop ecosystem, we had various frameworks for various data processing requirements. As a developer, you could use the MapReduce framework to analyze your data using your choice of programming languages such as Java, C++, Python etc. However, a data warehouse engineer who is also a SQL expert, has to learn one of these aforementioned programming languages to leverage the MapReduce framework. To overcome this problem, a new framework which runs on the top of Hadoop called “Hive” was introduced. There was a similar problem for ETL processing and so “Pig” was introduced. Similarly, tools like “Giraph” and “Mahout” were introduced for Graphs processing and Machine Learning respectively. Moreover, Hadoop is only used for batch processing and there was no way to process data in real time. So, for this a new framework called “Storm” was integrated with Hadoop to work with streaming data. With so many frameworks, organizations had a tough time maintaining all the frameworks and tracking the issues pertaining to them. Fortunately, all this would change with advent of Spark. As mentioned, Spark is a unifying platform which provides all these frameworks as one package with four major components.

Now, what actually does In-memory processing mean? Aren’t all the applications processed in memory only? Well, yes, all the applications are processed in-memory and written back to disk when processing is done, but Spark can process data in-memory and also retain the data within the memory or write to disk. Let us understand this with a figure by comparing Spark with MapReduce.

1(a) Data Processing with MapReduce

MapReduce Job

MapReduce Job

MapReduce Job

MapReduce Job

Read Write       Read       Write Read Write      Read      Write

Disk

Disk

Disk

Disk

In MapReduce, the data present in HDFS or any other Distributed file system is read by a MapReduce application and is processed in memory and then written back to disk after the job is complete. If the processed data is again needed for further processing, the data is again read from disk by a MapReduce application, processed in memory and then written back to disk. This process continues as per the requirement, as seen in the figure 1(a). The processes of reading and writing data from and to the disk increase the IO latency and so the overall job duration is increased. This is optimized in Spark as shown in the figure 1(b).

1(b) Data Processing with Spark

Spark

Memory

Spark

Memory

Read

Spark

Disk

Disk

Write

In Spark, the data is read from the disk, processed in-memory but instead of spilling it back to disk, Spark can retain the data within the memory for further processing. So, if the processed data is again required for further processing, the data is already present in the memory and the Spark application processes the data eliminating the IO latency and therefore the overall time to process the job is significantly reduced. With this, the processing speed when compared to MapReduce has been shown to be 100 times faster. The processed data from a Spark application can either be retained in memory or can be stored to the disk as per the requirement as shown in the figure 1(b).

The reasons, such as a unified platform for various data processing requirements and high speed in-memory processing, have gained worldwide popularity throughout the industry with almost all the major organizations using Spark for their data processing requirements.

Components of Spark

Now that we know why Spark is being used, let us dive in more and learn what Spark is made up of. Let us look in detail at each of Spark’s major components individually. The following figure 1(c) shows the components of Spark.

1(c) Components of Spark

Spark SQL

Spark Streaming

MLlib

GraphX

Spark Core

Scala

Python

Java

R

Spark Cluster Manager

YARN

Mesos

Kubernetes

Here is a brief explanation regarding these components so that we can better understand the Spark components.

Spark Core

Spark Core, as the name suggests is the core component of Spark which contains all the basic functionality for processing large datasets. Some of its functionality includes: managing memory, scheduling jobs, fault tolerance, using in-memory computation, referring datasets stored in storage systems etc. Spark Core includes a programming abstraction (API) called Resilient Distributed Datasets also known as RDDs, which is responsible for partitioning data across nodes on a cluster. With the help of these RDDs, the data can be transformed, collected and reduce things together. These RDD APIs can be referred by using any of the programming languages such as Scala, Python, Java and R as shown in the figure 1(c). We shall learn more about RDDs in the upcoming chapters.

Spark SQL

The Spark SQL component provides the developer with an SQL like interface to work with huge structured data which is distributed over a cluster of nodes. Spark SQL works well with structured and semi structured data. Spark SQL can also work with data sources such as Apache Hive tables, Avro, JDBC, ORC, JSON and Parquet file formats. Spark SQL also allows developers to combine RDD APIs along with Spark SQL code in a single application. We shall learn more about Spark SQL in the upcoming chapters.

Spark Streaming

The Spark Streaming component of spark deals with processing of real time data, also known as Streaming data. The streaming data can be from a fleet of web servers, sensors, IOT devices or any other sources which generate data. This enables Spark to ingest data as it is generated in real-time and perform data manipulation on that data. There are three major phases of Spark Streaming. They are  Gathering, Processing and Data Storage. Spark Streaming is also fault tolerant and scalable.

Spark MLlib

Spark MLlib is short for Machine Learning libraries which provides Machine Learning for huge datasets. MLlib contains various Machine Learning algorithms such as  Regression, Clustering, Classification and Collaborative Filtering. MLlib also contains lower level primitives such as generic gradient descent optimization algorithm. MLlib also uses the linear algebra package called Breeze for numerical computing. We shall be using Spark MLib to build recommender systems.

GraphX

GraphX deals with processing of Graphs in very efficient and distributed manner. GraphX extends the RDD APIs which allows a developer to create a directed multigraph with properties attached to each vertex and edge. This book does not cover GraphX.

Cluster Managers

Spark is all about processing massive amounts of datasets by distributing them over a number of nodes and scaling the cluster as required. To efficiently perform this task a cluster manager is required and Spark has its own cluster manager called  Standalone Scheduler. Spark can also be deployed using Hadoop YARN, Apache Mesos or Kubernetes as a cluster manager to schedule jobs and manage the resources of the cluster. We shall look into more about cluster managers in upcoming chapters.

Introduction to RDD

A Resilient Distributed Dataset (also known as an RDD) is the basic data structure of Spark, which is an immutable and fault tolerant collection of elements that can be computed and stored in parallel over a cluster of machines. Let us look at each key word individually and try to understand them in detail.

Resilient: The RDDs are fault tolerant to any data loss. Any loss in data due to hardware failure or data corruption can be recovered using the RDD lineage graph or DAG.

Distributed: The RDDs can be distributed over a cluster of machines in memory.

Dataset: The RDDs can be created with any of the datasets such as a text file, JSON, CSV, Database file via JDBC etc.

An RDD can be created in two ways. We can use the parallelize keyword on an existing collection or referencing data in an external storage system such as HDFS, shared storage system, HBASE or any other data source offering a Hadoop InputFormat.

There are two operations which can be applied on an RDD. They are Transformations and Actions. Transformations is the operation where an RDD is transformed into one or more RDDs while actions are the operations where a result is returned after computing an RDD.

Architecture of Spark

Spark is a unified parallel large-scale data processing engine. To achieve parallelism for processing huge data efficiently and with fault tolerance, the Spark framework was built. But how exactly does Spark work?

We must first understand the basics of data processing so that we can appreciate the importance of processing data in parallel. Consider processing data in your personal computer. You have a text file of a size ranging from a few MBs to a few hundred MBs. You can easily load that file to your computer and do various operations such as searching, sorting or filtering without any issues. Now, consider a plethora of text files each ranging from few hundreds of MBs to several hundred GBs or even a few hundreds of TBs. Let alone processing such data, it becomes harder to even save those files. This becomes a bottleneck for your computer. You now have two options. You can either increase the specifications of your computer or process a percentage of files in multiple computers. The first option is not feasible as you cannot keep on upgrading hardware to a single computer. The second option looks feasible. But how will you coordinate the processing on each computer to be efficient and accurate? How will you ensure that the data processed on one computer is not processed on another? How do you collect the results and merge them accurately? What happens if a computer stops working due to a hardware problem?

Apache Spark is the answer to all of those questions. Spark takes care of managing the group of computers/nodes called a cluster and coordinates the tasks on those nodes to process data. Spark uses a cluster manager to manage a cluster. There are various cluster managers such as Spark’s standalone cluster manager, YARN, Mesos or Kubernette. The cluster manager takes care of managing and coordinating the Spark applications.

As we know, Spark has a master-slave architecture. There is one master node and several worker nodes. The master node is simply called the Driver Program as it contains the program or the application which we develop to process the data. The Driver Program consists of an object called Spark Session, which provides access to all the underlying hardware for the Spark job. You can think of Spark Session as a gateway to access all the Spark functionalities similar to a database connection.

Driver

Spark Session

Driver Program

Cluster Manager

Executors

-Tasks

-Cache

-Tasks

-Cache

-Tasks

-Cache

-Tasks

-Cache

Prior to Spark 2.0, the Spark Context was the entry point or a gateway for a Spark application. The RDD was one of the main APIs of Spark which was accessed through the Spark Context. Similarly, to access other APIs other contexts were required. Developers had to create different contexts such as SQL Context for Spark SQL, Hive Context for Hive queries and Streaming Context for Spark Streaming. So, a level of abstraction called Spark Session was introduced which includes all the APIs mentioned above in a single object without the necessity to create multiple contexts. We shall look at these APIs in detail in the upcoming chapters.

The driver executes commands in the driver program while the cluster manager manages the resources on the cluster. As seen earlier, the cluster manager can be a Spark’s standalone cluster manager, YARN, Mesos or Kubernette.

The slaves are the worker nodes which are known as Executors in the Spark ecosystem. As the name suggests, executors are the work horses of Spark and are responsible for performing the tasks assigned to them by the Driver program. There can be any number of executors in a Spark cluster depending on the requirement. Since Spark is scalable, you can add more executors as your data increases.

The Spark session is responsible for breaking the job into smaller “chunks” and distributing the work to the executors while the cluster manager allocates resources for the Spark job on the executors. Once the executors finish processing the job, the cluster manager collects the processed data and reports it back to Spark.

Let us now look at a job workflow in Spark.

Job Workflow in Spark

The following is a workflow of a job in Spark.

●  The job is submitted by a client to the Spark cluster. The driver takes the code and converts the code to a Lineage Graph. This is the logical execution plan.

●  In this step, the Lineage Graph is converted to a physical execution plan or DAG where a bunch of tasks are created to run on the executors. All the optimizations to run the job are performed at this step.

●  The driver then contacts the cluster manager to allocate resources to run these tasks. The cluster manager launches executors on behalf of the driver. The driver then starts the tasks to the allotted executors based on the data placement. The executors send heartbeats to the driver while executing the tasks.

●  The driver monitors the progress of each task running on the executors. If a task fails on the executor, it is again restarted on another executor.

●  Once the executors complete their tasks, the result is sent back to the driver through the cluster manager.

Introduction to SparkSQL

What is Spark SQL?

Spark SQL is a library or module of Spark, which provides SQL style programming to process structured data. Spark SQL runs on top of Spark by wrapping all the Spark core APIs into a high-level abstraction. Spark SQL also provides optimizations to run the jobs faster (which is surprisingly lacking in Spark core), making Spark even more efficient. Since Spark SQL is syntactically similar to SQL, it is easier for developers, who already work on SQL, to become productive faster with less efforts. Spark SQL was implemented to overcome the disadvantages of running Apache Hive on top of Spark.

Why Spark SQL?

The following are the advantages of using Spark SQL.

●  Spark SQL is popular because it provides developers with easy to use APIs with support for various data sources. Spark SQL provides interfaces for programming languages and query languages which include SQL and HiveQL, helping developers to get productive in no time.

●  A wide variety of file formats such as csv, Avro, Json, Parquet, ORC etc. are supported by Spark SQL. It also supports almost all the relational databases with JDBC connectivity, which include MySQL, Postgress, and Oracle to name a few. NoSQL datastores such as HBase, Cassandra, and Elasticsearch are also supported with Spark SQL.

●  Spark SQL can also be easily integrated to other Spark libraries, which include Spark ML, GraphX and Spark Streaming.

●  Spark SQL efficiently processes structured data by advanced optimization techniques such as cost based optimizer, in-memory columnar caching, code generation and reduced disk IO.

Spark SQL Architecture

The Spark SQL Architecture consists of the following components.

DataSource API

The Data Source API is the universalAPI to load and store structured data. This is similar to the textFile, binaryFile, Sequence File APIs in Spark core (RDD). Instead of so many different APIs for different formats of data, we now have the DataSource API which can load and store structured data. The DataSource API has built-in support for JDBC, Hive, Avro, JSON, Parquet, etc. The DataSource API can automatically infer schemas without the user explicitly mentioning the schema. We can also specify the schema using Data Source API.

Data Frames

Data Frames are like advanced version of RDDs. Data Frames are distributed collection of data represented in the form of rows and named columns. All the features of RDDs also apply to Data Frames. They are distributed, lazy, can be cached and are immutable. In other words, Data Frames are similar to that of tables in RDBMS but with even more advanced capabilities. Since Data Frames are similar to that of RDBMS tables, we can simply run SQL like queries on our Data Frames, and have the data processed on our Spark cluster in distributed manner.

SQL Interpreter & Optimizer

The queries on Data Frames are run in SQL which is a high-level  language. So, we need a SQL Interpreter or a SQL Parser which will interpret or parse our SQL queries. The Optimizer in Spark SQL is called a Catalyst. The Catalyst optimizer works on the SQL Data Structure trees and transforms the logical plan to an optimized plan, which in turn will be transformed to a physical plan. In simple terms, this component helps us process our big data in an efficient and optimized way.

Spark SQL Thrift Server

The Spark SQL Thrift server is used as an abstraction layer to connect Spark SQL with various Business Intelligence (BI) tools. The Spark Thrift Server is similar to that of Hive Thrift Server. Instead of running queries from BI tools via Hive Thrift server as Map Reduce jobs, we can use the Spark Thrift Server and use Spark’s features. Since Spark is faster than Hadoop Map Reduce, we can have our queries processed faster in and in a more efficient manner. The BI tools can be connected with Spark using the JDBC or ODBC drivers. For example, Tableau can be connected to Spark SQL using the ODBC driver and then run the queries from Tableau on Spark.

Tungsten

Tungsten is a Spark SQL component which helps in the memory tuning and optimization of Spark jobs. Tungsten’s memory management is developed to address the drawbacks in the JVM’s memory management modules. With Catalyst and Tungsten the Spark SQL jobs are much faster and efficient when compared to RDDs.

What are Datasets?

A Dataset is the most advanced API in Spark. Datasets are an extension of DataFrames, which overcomes all of the disadvantages of both RDDs and DataFrames. The Dataset API provides developers with a type safe mechanism as well as a functional style programming option, while retaining the relational type of programming in DataFrames and performance optimizations. Hence, it is an extension of DataFrames. Datasets were introduced in Spark 1.6.

Datasets use Encoders to serialize and deserialize the Spark SQL representation of JVM objects. The serialization and deserialization with encoders are significantly faster when compared to Java serialization. In simple words, datasets use encoders to convert the data between JVM objects and Spark SQL representation of tabular objects.

Why Datasets?

The following are the advantages of using datasets.

●  Datasets are a combination of RDDs and DataFrames. Datasets help you code like you would with RDDs and process them through performance optimization engines.

●  Datasets have all the other features as RDD and DataFrames. They are lazy, immutable, distributed and can be cached.

●  Datasets assures type safety similar to that of RDD. Any type errors are flagged at compile time rather than being notified at runtime.

●  Data processing with datasets is optimized using the Catalyst optimizer and Tungsten similar to that of DataFrames. This ensures very fast and efficient processing of data.

●  The Dataset API provides the developers with a functional style of programming as well as relational style of programming.

●  Datasets can process structured as well as unstructured data. Datasets can also automatically infer schema.

●  Datasets can be converted from RDD and DataFrames.

Spark Data Storage

Spark supports major file systems such as HDFS, Amazon S3, Azure Blob etc. Spark supports the local file system for storing the data as well. However, using a distributed file system such as HDFS can leverage the power of Spark by distributing the datasets throughout the cluster. Spark is also capable of dealing with various file formats such as text, ORC, parquet etc. which we shall be covering in detail in the upcoming chapters.

Various Spark Versions

Apache Spark is an open source Apache Software Foundation Project which follows semantic versioning guidelines with a few deviations. All the Spark releases are versioned as [MAJOR].[FEATURE].[MAINTENANCE] for example Apache Spark 1.2.1 Which means Spark has a major version of 1 with feature version as 2 and Maintenance version as 3. Please check the link in references section for more about versioning in Spark.

The major versions of Spark are considered as Apache Spark 1.0 and Apache Spark 2.0. However, we shall be using Apache Spark 2.4.1 throughout the book as this is the latest version of Apache Spark at the time of writing this book.

Introduction to Artificial Intelligence

What is AI?

As quoted by John McCarthy, one of the founders of the discipline of artificial intelligence:

“Artificial Intelligence is the science and engineering of making intelligent machines, especially intelligent computer programs”.

AI is an approach to make computers solve problems by learning from their previous mistakes. Simply put, artificial intelligence aims to make computers work like a human brain and perform intellectual tasks similar to humans. The purpose of AI is to improve computer functions which involve intelligence. Intelligence can be defined as reasoning, learning, problem solving, perception and linguistic intelligence.

The applications of Artificial Intelligence include Natural Language Processing (NLP), gaming, speech recognition, handwriting recognition, robotics, vision systems etc. Artificial Intelligence is a broad concept which simulates human intelligence while Machine Learning is a subset of AI and deep learning is a subset of Machine Learning. They key is that while AI focuses on simulating intelligence, machine learning focuses on actual learning.

What is Machine Learning?

Machine Learning is a subset of Artificial intelligence which enables computers/machines to learn and identify patterns from historical data and make predictions on new data using algorithms. Once an algorithm is trained, we call it a model. Deep Learning is a subset of Machine Learning. Deep Learning is inspired by the structure and function of the brain and is also called an artificial neural networks.

The applications of Machine Learning include recommender systems, search algorithms and many more. Machine learning works with structured and semi structured data. Machine learning is classified into three types.

●  Supervised Learning

●  Unsupervised Learning

●  Reinforcement Learning

The machine learning techniques are categorized in three common types based on the required output.

●  Classification

●  Clustering

●  Collaborative Filtering

We shall be looking at these categories in detail in the next chapter. We shall also be looking at the capabilities of Spark MLLib in the next chapter.

What is Deep Learning?

Deep learning is comprised of several layers and hidden layers of neural networks between the input data and output prediction. Deep Learning uses large amounts of data to learn the patterns, classify information into labels and categories and make decisions accurately.

Deep Learning applications include self-driving cars, virtual assistants, fraud detection, visual recognition etc.

Artificial Intelligence, Machine Learning and Deep Learning can be represented as shown in the following picture.

Artificial Intelligence

Machine Learning

Deep Learning

That’s all the theory for this chapter. Let’s move ahead to our lab exercises and install Spark and IntelliJ to run our programs.

AIM

The aim of the following lab exercises is to install Spark, Scala and the IntelliJ IDE.

The labs for this chapter include the following exercises.

●  Downloading and Installing JDK

●  Downloading and Installing Scala

●  Downloading and Installing Spark

●  Downloading and Installing IntelliJ IDEA

●  Configuring IntelliJ IDEA

We need the following packages to perform the lab exercise:

●  Java Development Kit

●  Scala

●  Spark

LAB EXERCISE 1: Installations

TASKS:

1. Download and install JDK

2. Download and install Scala

3. Download and install Spark

4. Download and Install IntelliJ IDEA

5. Configuring IntelliJ IDEA

Task 1: Download and Install JDK

Step 1: From the terminal, run the following commands to install the JDK (Java Development Kit).

$ sudo apt-get update

This will update the package index in linux. You might be asked to enter your password after you run the update command.

Step 2: Once you run the update command, run the following command to actually download and install JDK.

$ sudo apt-get install default-jdk

The prompt will ask you to hit ‘Y’ after running the above command as shown in the screenshot. Hit ‘Y’ from your keyboard to continue with the installation and then hit the Enter key. This will download and install the JDK on your machine.

The installation process might take some time depending on your internet connection. Please allow it to download and install completely. You should see the following message when the installation is complete. (Obviously you may see a different version of Java because the “default” install may change).

Step 3: Run the following command to validate that Java has been installed successfully. The terminal should show the Java version (similar to what is shown in the screenshot, again – your version may differ).

$ java –version

Task 1 is complete!

Task 2: Download and Install Scala

Now that we have installed Java, we are ready to install Scala and start writing some Scala code!

Step 1: Run the following command from the terminal to install Scala.

$ sudo apt-get install scala

The prompt will ask you to hit ‘Y’ after running the above command as shown in the screenshot. Hit ‘Y’ from your keyboard to continue with the installation and finally hit the Enter key.

Step 2: Verify your Scala installation version by running the following command.

$ scala -version

Step 3: After the installation is completed successfully, type scala in your terminal and you will be prompted with a Scala prompt.

$ scala

This completes the Scala installation. The scala prompt is the interactive shell where you can write and run Scala code. This interactive shell is also known as a REPL (Read Evaluate Print Loop).

Step 4: You can now start writing Scala code! Let’s start by printing the classic “Hello world!” from the shell. To do this simply type the following code and hit enter on your keyboard.

scala> println(“Hello World!”)

As you can see from the screenshot, the output is shown below immediately as soon as you hit enter.

Step 5: To quit the Scala REPL, you use the following command.

scala> :q

You are now back to the terminal prompt.

Task 2 is complete!

Task 3: Download and Install Spark

Note: Java Development Kit (JDK) is a prerequisite to install Spark. Make sure you have installed it if you have not performed the Lab Exercise 1. Please follow the steps in task 1 of Lab Exercise 1 and then come back here to Install Spark.

Step 1: Verify if your machine has Java installed by executing the following command as shown below.

$ java –version

You will see the version displayed in the terminal as shown in the screenshot above. If you do not see the Java version displayed, your machine does not have Java installed. Please install JDK before continuing to next step.

Step 2: Let us install Spark in Standalone mode. Navigate to the download URL below, select the latest stable version for Spark (which is Spark 2.4.2 at the time of writing this book) and select the package type which is Pre-built for Apache Hadoop 2.7 and later. After selecting the version click the link as shown in the screenshot below to download Spark.

Download URL: http://spark.apache.org/downloads.html

After clicking the download link, you will be taken to a page with mirror site to download Spark. Click the mirror link as shown below and your download should start. The download may take a while depending upon your internet connection.

Step 3: The download will be saved to the Downloads directory by default but this could be different based on your machine configuration. Execute the following command to change the directory to Downloads folder.

$ cd Downloads

Once you are in the Downloads directory, you may optionally check if Spark has been downloaded using the ls command.

$ ls

Now that you are sure that you have the Spark tar file, untar the Spark tar file to /usr/share directory using the command below.

$ sudo tar –xvf spark-2.4.2-bin-hadoop2.7.tgz –C /usr/share

The file will start to untar to /usr/share directory as shown in the screenshot above. You can verify the same by executing the following command below.

$ cd /usr/share

$ ls sp*

As we can see from the screenshot above, the Spark directory is listed.

Let us make a soft link to the Spark directory so that we don’t have to refer to Spark with entire name as above. This will also be useful for future updates. Execute the following command.

$ ln –s spark-2.4.2-bin-hadoop2.7 spark

Run the following command again to check if we were able to create the soft link successfully.

Step 4: Let us now set up the environment variables for Spark. Execute the following command to do so.

$ sudo vi ~/.bashrc

The file should open as shown below.

Now press i key to edit the file and append the following environment variable at the end of the file.

SPARK_HOME=/usr/share/spark

Export PATH=$SPARK_HOME/bin:$PATH

After you have finished appending the text above, hit the Esc on your keyboard to stop editing and then press Shift - Z - Z to exit out of the editor by saving the changes. (Please see that you need to press Z twice while holding Shift key.)

Now reload the modified .bashrc file using the following command.

$ source ~/.bashrc

Step 5: Let us now test the Spark installation by accessing the Spark Shell. To do so, execute the following command.

$ spark-shell

Once you run the above command, a screen similar to the screenshot shows confirming the successful installation of Spark.

Task 3 is complete!

Task 4: Download and Install IntelliJ IDEA

So far, we have been writing our code in the Spark Shell. Spark Shell is great for developing applications with a small set of data. But in the real world, you will be using an integrated development environment also known as IDE for developing Software applications.

We shall be using IntelliJ IDEA to write and run Spark-Scala code throughout the lab exercises. In this task we shall download and install IntelliJ IDEA and then install Scala plugin so that we can write Scala code. It is important to note that there are several, equally viable IDEs that we can use that do as good of a job or better than IntelliJ such us: Zeppelin, Theia, VSCode, and others.

Step 1: Navigate to the following URL from your web browser and click on the “Download” button for Community edition as shown in the screenshot below.

http://bit.ly/2V1HFYO

The file should begin to download automatically to the Downloads folder. The download might take a while depending upon your internet connection.

Step 2: Once the download is complete, open the terminal and run the following commands to untar. We shall be extracting the tar ball to the /opt directory.

$ sudo tar –xzf Downloads/ideaIC-2019.1.1.tar.gz –C /opt

Please note that your version of IntelliJ might be different. Please replace the version in all commands with the version you downloaded.

Step 3: Now run the following command to install IntelliJ.

$/opt/idea-IC-191.6707.61/bin/idea.sh

Please note that your path or version might be different.

You should now see a prompt asking to import settings. Simply click “OK”.

Step 4: You should now be prompted with a Privacy Policy window. Click on the check box to accept the policy and click on “Continue” button as shown in the screenshot.

In the next prompt, click on “Don’t Send”.

Step 5: You will now be prompted to select a theme. Please select a theme you are comfortable with and click on the “Skip remaining and Set Defaults” button as shown in the screenshot.

You should now see the welcome screen as shown below.

This concludes the installation of IntelliJ IDEA. But to run Spark Scala code, we need to install the Scala plugin.

Step 6: Click on the “Configure” button as shown in the screenshot and click on “Plugins” in the dropdown menu.

Step 7: The marketplace for plugins will be opened. Click on the “Install” button for “Scala” plugin as shown in the screenshot. If you do not see “Scala”, search for “Scala” in the search bar above and then click the “Install” button.

This action should begin the download. Once it is downloaded, you will be asked to Restart the IDE. Please click on “Restart IDE” button and the confirmation.

The IDE will now restart and show the welcome screen again. With this, you have successfully installed IntelliJ IDEA with the Scala plugin.

Task 4 is complete!

Task 5: Configuring IntelliJ IDEA

Step 1: Click on the “Create New Project” button on the welcome screen.

Step 2: You will then be taken to “New Project” screen. Click on Scala in the left panel, select “SBT” and then click “Next” as shown in the screenshot below.

Step 3: After clicking “Next” in the previous step, you will be taken to prompt to enter project name. Enter project name as “Spark”. Make sure the JDK, SBT and Scala versions are selected automatically as shown in the screenshot. Also, check the “Sources” checkbox for both SBT and Scala if not already checked. Finally click the Finish button.

Step 4: You will now be taken to the IDE interface. Click on the “Project” as shown in the screenshot if not already open.

Expand the Spark project by clicking on the small triangle to the left, if not already expanded and double click on build.sbt file as shown in the screenshot.

Step 5: Now go to the Maven Repository for Spark using the URL below.

http://bit.ly/2UQvnNU

Click on the Spark Project Core link as shown in the screenshot.

Next, select the version of Spark you have installed. We have installed Spark 2.4.2 in this book and hence we will be clicking on 2.4.2 link for Spark. Please select the correct version as per your installation.

Now, select SBT tab and copy all the lines of code for SBT and paste it in the build.sbt file.

Go back to the Maven Repository page and copy paste the Spark Project SQL and Spark Project ML Library libraries as well in the build.sbt file. Finally, click on the “Import Changes” to finish configuration.

You can add new libraries if required by following the same procedure.

You are now ready to write your first Spark program!

Task 6 is complete!

SUMMARY

Apache Spark is an open source, fast, unified, parallel, large-scale data processing engine. Spark provides a single platform for various data processing tasks such as Big Data analytics, Streaming, Machine Learning and Graph processing. Spark is up to 100 times faster than MapReduce due to its In-Memory processing and lazy execution of tasks. Apache Spark has its own built-in cluster manager called “Standalone Scheduler” or can be deployed on “Hadoop YARN”, “Apache Mesos” or “Kubernette”.

AI is an approach trains computers to solve problems by learning from their previous mistakes. Simply put, artificial intelligence provides code that allows machines to simulate human intelligence. The purpose of AI is to improve computer functions which involve intelligence. Intelligence can be defined as Reasoning, Learning, Problem Solving, Perception and linguistic intelligence.

REFERENCES

●  http://spark.apache.org/

●  http://spark.apache.org/docs/latest/

●  http://hadoop.apache.org/

●  http://spark.apache.org/versioning-policy.html

●  https://spark.apache.org/docs/latest/ml-guide.html

CHAPTER 2: RECOMMENDER SYSTEMS USING APACHE SPARK

Theory

In the previous chapter we had a brief look at the history of Big Data with an introduction to Hadoop and MapReduce. We also looked at Spark and Artificial Intelligence. Let us go further in this chapter and look at Machine Learning in detail and build a recommender systems using Apache Spark.

Basics of Machine Learning

As seen in the previous chapter, Machine Learning is classified into three types.

●  Supervised Learning: Supervised learning is a type of Machine Learning, where the algorithm is trained with example data that is all labeled. The example data contains both the input data and the output results from previous observations. The algorithm learns over time using the labelled data and predicts the output when new data is introduced. Supervised learning is task oriented. The more labelled example data that is provided to the algorithm, the more accurate it gets in predicting the output. Supervised learning is also known as predictive analysis.

Supervised Learning techniques include:

- Classification

- Regression

Let us look at a simple example of how supervised learning works. An algorithm is fed with images of dogs as input and the label dog as output. When a new image of a dog is provided as input, the algorithm will correctly predict the output as dog.

●  Unsupervised Learning: Unlike supervised learning, the unsupervised learning algorithms are not provided with labelled data as examples to learn. The unsupervised learning algorithms learn from the data by identifying patterns based on similarity, density, regularity, structure etc. An example of unsupervised learning is grouping similar items using clustering.

The unsupervised learning techniques include

- Clustering

- Collaborative Filtering

- Frequent Pattern Matching

●  Reinforcement Learning: Reinforced learning is a bit different from supervised and unsupervised learning. Reinforcement learning is a type of learning where you get rewarded for making good decisions and punished for making a bad decisions. The rewards encourage the algorithm to make better choices as time goes on. Eventually, with reinforcement learning, the machine learns to make the best decisions in every situation. Reinforcement learning is learning from mistakes and getting better with time.

Reinforced Learning is a machine learning technique which allows an agent to make decisions or perform actions in an environment that maximize the accuracy (rewards).

Machine Learning Techniques

In the previous chapter we looked at common machine learning techniques. Let us now look at them in detail.

●  Classification: Classification is one of the most widely used supervised machine learning techniques in which the algorithm learns to classify new data, based on a set of known categories or labels. Examples of classification include: filtering emails by spam and not spam, tagging transactions as authorized or fraudulent.

●  Clustering: Clustering is an unsupervised machine learning technique where similar data is grouped into similar clusters. An example of clustering is to identify fake news based on the content.

●  Collaborative Filtering: Collaborative Filtering is an unsupervised machine learning technique that can be used to predict which items a user might like or buy based on the previous reactions of other users. Collaborative Filtering can be used in recommender systems as the data is filtered based on the recommendation of others.

Getting Started with Recommender Systems

Now that we have basic knowledge about Machine Learning, let us get started with Recommender Systems.

What is Recommender System?

Recommender systems are the systems which have the ability to predict similar things when a user selects or does something online. E-Commerce websites such as Amazon use recommender systems to predict and suggest products similar to the products purchased or viewed by the customer. Video and music streaming platforms such as Netflix, YouTube, Spotify and Pandora use recommender systems to suggest video and music playlists based on the user’s previous streaming history, likes and dislikes or ratings. Social media platforms such as Facebook and Twitter use recommender systems to suggest content to their users that are most relevant to them. A large amount of wealth is being invested by various industries to build powerful recommender systems for further improving user experience and there by sales.

Why Recommender Systems?

Why do we need recommender systems? A huge amount of data is being generated every day. There is a lot of content and tons of products that are available for users to choose from. It is because there are so many options available for people to choose from, recommender systems help them select the items they actually want. For example, YouTube has the largest collection of videos. With recommender systems, users are suggested the videos they would actually like watching based on their previous watch history.

This is the requirement today because the internet has so many options to choose from. Recommender systems were not required a couple of decades ago as the internet was not common and people used to shop in physical stores. The products available were limited in the stores based on the size of the store. In other words, the data, techniques, and technology was not available to generate recommenders.

Recommender Systems Techniques

There are two basic types of techniques or algorithms when it comes to building recommender systems.

●  Content-Based Filtering

●  Collaborative Filtering

Let us now look at these algorithms in detail.

Content-Based Filtering

Content based filtering, also known as cognitive filtering, provides recommendations by comparing similarities between item features with the items the user already likes based on the user’s actions or through a feedback mechanism such as ratings. The items here could be products, movies, songs – really anything. Let us understand this concept better by listing TV shows and comparing their features.

Friends

The Big Bang Theory

Game of Thrones

The Witcher

Comedy

Yes

Yes

No

No

Fantasy

No

No

Yes

Yes

Emmy Award

Yes

Yes

Yes

No

Warner Bros

Yes

Yes

No

No

Netflix

No

Yes

No

Yes

The table above lists features in the first column and the rest of the columns provide the information regarding the features. The features listed for every TV show specify if the show is a genre of comedy or fantasy, has received an Emmy award, is produced by Warner Bros and its availability on Netflix. The table contains all yes and no values but the values could be any arbitrary values.

For content-based filtering, we consider two columns and compare how similar they are. For example, if a person has liked Friends, what other shows might they be interested in? To answer this, we need to take all the features of the Friends series and compare them with features of other shows listed in the table. Let us compare Friends with The Big Bang Theory. Both these shows belong to comedy and not fantasy genre. They both have won Emmy awards and both are produced by Warner Bros. So, the match score between these two TV shows is 4. Now, if we compare Friends with Game of Thrones, their match score is 2. Similarly, the comparison between Friends and The Witcher is 0.

Based on the match score, we can conclude that a person who likes Friends might also like The Big Bang Theory and so this would be a good recommendation to make. These content-based filtering systems worked based on the attributes of the TV shows. We can build these systems by knowing about the products we have and use any features of those products.

Let me take a second to point out that this is math, not magic. The user may or may not like the recommendation. If they do not like the recommendation, the model can recognize this and adjust for future recommendations.

Collaborative Filtering

Collaborative filtering is an unsupervised machine learning technique used to build recommender systems, by predicting the items a user might like or buy based on the reactions of other users. Unlike content-based Filtering, collaborative filtering relies on how other users have responded to the same items and not on the features of the item itself.

Let us understand this concept by again looking at the TV shows we have seen in the previous section. However, instead of having features of the TV shows, let us list a bunch of people who like or dislike a particular TV show and then recommend a TV show to a user based on the likes and dislikes of other users. There could be various methods to get this feedback (likes or dislikes for example). To generate this data, we can ask the users for their feedback directly by a method of surveying (e.g.: user ratings) or by implicitly recording their actions such as which show they choose to watch.

Friends

The Big Bang Theory

Game of Thrones

The Witcher

Ernesto

Yes

Yes

Yes

Yes

Mathew

No

Yes

No

Yes

Uzair

Yes

No

Yes

No

Zayyan

No

No

Yes

Yes

From the table above, we can depict that Ernesto has liked all the shows they watched. Mathew liked The Big Bang Theory and The Witcher while Uzair liked the opposite of Mathew, i.e., Friends and Game of Thrones. Zayyan liked Game of Thrones and The Witcher; none of the comedy shows.

Based on this data, how do we recommend a TV show to a new individual if they already like one of the shows above? Let’s say the new individual is John, who likes the TV show, Friends. Which other show would John also want to see based on what similar people did? To answer this question, we should filter the list to only consider the people who liked Friends. Since Mathew and Zayyan do not like Friends, they are filtered out and not taken into consideration for the recommendation. From the table above, we can see that Ernesto and Uzair liked Friends. Now, we go ahead and check which other shows Ernesto and Uzair like. Let us count the number of likes for each show liked by these two users. The Big Bang Theory has 1 like, Game of Thrones has 2 likes and The Witcher has 1 like.

So, we want to recommend a show to John which they are likely to watch and based on our algorithmic logic, we recommend Game of Thrones. Again, this is based on the number of likes from the other users who also like Friends. This recommendation is based on the likes of other people who also like the show John likes. Hence the name Collaborative Filtering.

This technique only works when the data about the users’ preferences is already available. There has to be a reasonable quantity of data in order to make recommendations using Collaborative Filtering. The more data we have the better recommendations we can make.

Spark MLlib Overview

Apache Spark provides a Machine Learning library called Spark MLlib. Spark MLlib is built on top of Spark. It consists of many popular Machine Learning libraries or algorithms and utilities designed for simplicity, scalability, and easy integration with other tools. The algorithms and utilities in Spark MLlib include classification, regression, clustering, collaborative filtering, Linear Algebra and Statistics. Spark MLlib is perfectly integrated with other components of Spark such as Spark SQL, DataFrames and Spark Streaming.

The following are the benefits of Spark MLlib.

●  Spark MLlib provides scalability. The same Machine Learning code can scale from one machine to a cluster of machines without any problems.

●  With MLlib on top of Spark, we can build an end to end machine learning model without worrying about deploying multiple tools to ingest (Spark Streaming) and process (Spark SQL and DataFrames) data. Spark also provides graph computations using Spark GraphX.

●  Spark MLlib provides simple and user-friendly APIs, so that the users who use different tools such as Python and R can easily run algorithms without any issues.

●  Spark MLlib can use the performance and efficiency of Spark for quick processing.

Spark MLlib Components

Spark MLlib consists of the following components.

Algorithms

Classification

Regression

Clustering

Collaborative Filtering

Pipeline

Constructing

Evaluating

Tuning

Persistence

Utilities

Linear Algebra

Statistics

Data Handling

Featurization

Extraction

Transformation

Dimensionality reduction

Selection

We have already seen the explanation of the algorithms in the previous sections.

Pipeline

A pipeline in Spark MLlib is a high-level API that consists of a sequence of stages for running a machine learning workflow. The main stages of a pipeline are Transformer and Estimator. The following shows the concept of pipeline in MLlib.

Load Data

Transformer

Model Evaluation

Estimator

●  Transformer: The Transformer is an algorithm which consists of a transform() function that maps one dataframe to another. The examples of transformer include Normalizing the data, tokenizing sentences to words, converting categorical values to numbers etc. There can be any number of transformer stages depending upon the requirement.

●  Estimator: The Estimator is an algorithm which consists of a fit() function that maps DataFrames into a model. The Estimator is a learning algorithm that trains on data and returns a model, which is a transformer. There is only one estimator stage at the end of the pipeline. For example, a learning algorithm such as LogisticRegression is an Estimator, and calling fit() trains a LogisticRegressionModel, which is a Model and hence a Transformer. Estimator = Algorithm. Transformer = Final Model. A Final Model is a trained Algorithm.

Alternating Least Squares

Spark MLlib implements the Alternating Least Squares (ALS) algorithm for collaborative filtering to make recommendation systems. Recall the table above in the Collaborative Filtering section. Let us call it user matrix table, where we have ratings of users and we want to build a recommender system for a new user based on the ratings of these users. Since the user has rated one of the movies, we should recommend a new movie to the user based on that rating and the ratings of movies rated by similar users. The new movie we would like to predict is the missing entry in the user matrix table. This missing entry can be predicted by small set of latent factors.

The ALS implementation in Spark MLlib has the following parameters.

●  The rank parameter specifies the number of latent features to be used.

●  The iterations parameter specifies the number of iterations ALS has to run. The reasonable solution is to have the number of iterations at 20 or less.

●  numBlocks is the number of blocks used to parallelize computation (set to -1 to auto-configure).

●  lambda specifies the regularization parameter in ALS.

●  implicitPrefs specifies whether to use the explicit feedback ALS variant or one adapted for implicit feedback data.

●  alpha is a parameter that is applicable to the implicit feedback variant of ALS that governs the baseline confidence in preference observations.

The default values are as follows: numBlocks: -1, rank: 10, iterations: 10, lambda: 0.01, implicitPrefs: false, and alpha: 1.0.

Please check the references section for more information on ALS.

That is all the theory for this chapter. Let is now proceed to the labs and build a movie recommender system using Apache Spark.

AIM

The aim of the following lab exercises is to build a movie recommender system using Apache Spark. This exercise requires the Scala programming language as a pre-requisite.

The labs for this chapter include the following exercises.

●  Download MovieLens Dataset

●  Create a New Package in IntelliJ IDEA

●  Mapping Movie Ids with Movie Names

●  Loading and Mapping User Data

●  Training Recommendation Model

●  Running the Model

We need the following packages to perform the lab exercise:

●  Java Development Kit

●  Scala

●  Spark

LAB EXERCISE 2: Movie Recommender System using Apache Spark

Tasks

1. Download MovieLens Dataset

2. Create a new package in IntelliJ IDEA

3. Mapping Movie Ids with Movie Names

4. Loading and Mapping User Data

5. Training Recommendation Model

6. Running the Model and Making Predictions

We are now going to build a movie recommender system using Apache Spark. Before we begin, we need to first download the data required to build the recommender system.

Task 1: Download MovieLens dataset

Step 1: Navigate to the following link and download the MovieLens 100K Dataset.

MovieLens 100K Dataset -https://grouplens.org/datasets/movielens/100k/

Step 2: Extract this zip file and you should see several files. We shall be using u.data and u.item files for this exercise. Please save the files in a folder where you can refer them to when required. This is real data provided by real users.

The u.data files consists of movie ratings given by users with four tab separated columns: userId, movieId, rating and timestamp.

The u.item file consists of information about the movies with a total of 19 ‘|’ seperated columns. The columns are movieId, movie_title, release_date, video_release_date, IMDb_URL, unknown, Action, Adventure, Animation, Children's, Comedy, Crime, Documentary, Drama, Fantasy, Film_Noir, Horror, Musical, Mystery, Romance, Sci-Fi, Thriller, War, Western.

Step 3: We will insert a couple of ratings by adding a new user in u.data file. We shall then recommend movies to this new user, based on likes of similar users who also rated the movies this new user liked.

Open the file u.data and add the following three ratings as shown in the screenshot below. We have added three ratings for the new user 944. The first two movies are The Mask and Dumb & Dumber. They both are rated 5 stars and both belong to comedy genre. The third movie is The Devil’s advocate belonging to horror/thriller genre. We have rated it as 1. Remember the column definitions: userID, movieID, Rating, and TimeStamp.

We have created a profile of user who likes comedy movies and doesn’t like horror movies. We then feed this data to our recommender system to recommend movies to the new user based on their ratings by comparing the ratings of other users, who have also rated the movies in a similar way.

Task 1 is complete!

Task 2: Creating a new package in IntelliJ IDEA

Step 1: Open IntelliJ IDEA, which was installed in the previous exercise. If you cannot find it, click on the Apps menu on your desktop as shown in the screenshot.

Now, click on the IntelliJ IDEA icon as shown in the screenshot. You can also search for IntelliJ IDEA if you are unable to find it in the Apps menu. Once IntelliJ IDEA is loaded, right click on the icon which is present in the left task bar and click on “Add to Favorites”. This way, you can quickly access when you need it the next time instead of repeating all this process over and over again.

You should see an interface as shown in the screenshot below.

Step 2: Expand the src folder by clicking on the caret. You should now see a main folder. Expand it as well and you should see a folder called scala.

Step 3: Now, right-click on scala folder and hover over New and click on Package as shown in the screenshot below.

You will be asked to enter a name for your package. You may enter any name you like. We have named our package training. After entering the name, click on OK to finish creating a package. You should now see a package below the scala folder as shown in the screenshot.

Task 2 is complete!

Task 3: Mapping Movie Ids with Movie Names

Step 1: Now that we have our dataset ready, let us start building our movie recommender system. Right-click the package which you have created in previous step and hover over New and then click on Scala Class.

Running the Model

Step 2: You will be prompted with a pop up asking you to enter the name of your Scala class. Please enter movieRecommender and then click on the dropdown for Kind and select Object.

Step 3: Click OK and you should see a new tab open on the right side and also with the Scala object in left panel as shown in the screenshot below.

You will see that the name of the package and the object should be prepopulated in the IDE.

Step 4: We are now ready to start writing the code for recommender system. The first thing we need to do is to include the import statements from the Spark libraries. This program requires the following imports.

import org.apache.spark._

import org.apache.spark.SparkContext._

import scala.io.Source

import scala.io.Codec

import java.nio.charset.CodingErrorAction

import org.apache.spark.mllib.recommendation._

The first two import statements are used to import the Spark packages. The next three import statements that are Scala specific imports which are used to read files from file system and correct any errors in the character encoding. The last import statement is to import the Spark MLlib recommendation algorithms.

Please make sure you enter the import statements above the object line as shown in the screenshot above.

Step 5: Now that we have the required imports, let us now define a function which would load the movie names to a Map object.

def loadMovieNames() : Map[Int, String] = {

The function loadMovieNames() does not take any arguments and returns a Map object of Ints and Strings.

Step 6: Let us now handle character encoding issues. This will make sure any malformed records are handled accordingly.

implicit val codec = Codec("UTF-8")

codec.onMalformedInput(CodingErrorAction.REPLACE)

codec.onUnmappableCharacter(CodingErrorAction.REPLACE)

Step 7: Now, let us declare a movieNames variable of type Map which maps Int to String and initialize it as an empty map.

var movieNames:Map[Int, String] = Map()

Next, we load the data from our file using Source.fromFile method and call getlines method to get each line based on /n character. Please make sure you refer the correct path to file u.item. Next, we iterate through each record in our input data using the for comprehension, and split each field based on ‘|’ as we know our fields are delimited by a ‘\’. Next, we check and ensure that each record has two fields. Next, map the movie Id with the movie name, by adding the fields to the movieNames Map object. Finally, return the Map object as required by our function.

val lines = Source.fromFile("RSAS/ml-100k/u.item").getLines()

for (line <- lines) {

var fields = line.split('|')

if (fields.length > 1) {

movieNames += (fields(0).toInt -> fields(1))

}

}

movieNames

}

Your code should now look similar to this.

Task 3 is complete!

Task 4: Loading and Mapping User Data

Step 1: Now that we have the movie ids mapped to the movie names, let us start coding the main function for the recommender system. Write the main function and create a SparkContext object as shown below.

def main(args: Array[String]) {

val sc = new SparkContext("local[*]", "movieRecommender")

Step 2: Now, load the movie names from the map object to a variable movieDictionary by assigning it to the loadMovieNames() function.

val movieDictionary = loadMovieNames()

Step 3: Now that we have the movie names loaded into a variable, let us get the u.data file by using Spark’s textFile method and split the records by tab character as shown below.

val data = sc.textFile("RSAS/ml-100k/u.data")

val ratings = data.map( x => x.split('\t') )

Step 4: Next, let us assign the split fields to a Rating RDD as shown below. The first three fields in u.data file (movieId, userId and rating) are mapped to a Rating RDD. We also use the cache() method to cache the splitRatings object as the program will be using this object more than once.

val splitRatings = ratings.map( x => Rating(x(0).toInt, x(1).toInt, x(2).toDouble) ).cache()

Your program should now look like the screenshot below.

Task 4 is complete!

Task 5: Training Recommendation Model

Step 1: Let us now start training our recommendation model by assigning the rank and iterations parameter to 8 and 20 respectively. There is no rule of thumb on how to set these parameters. However, the above are the recommended optimum values for these parameters based on training experiments.

val rank = 8

val iterations = 20

Step 2: The next step is to simply call the train method by passing the arguments splitRatings, rank and iterations on ALS as shown below. This is where the magic happens. The computer is “learning”.

val model = ALS.train(splitRatings, rank, iterations)

Step 3: Now, let us specify the user id we need the recommendations for and print the ratings given by this specific user to the console. We use the filter method to filter out the ratings given by the user. Next, we print the ratings to the console with the movie name and ratings. In this case, we need the recommendations for user id 944.

val userID = 944

println("\nRatings for user ID " + userID + ":")

val userRatings = splitRatings.filter(x => x.user == userID)

val myRatings = userRatings.collect()

for (rating <- myRatings) {

println(movieDictionary(rating.product.toInt) + ": " + rating.rating.toString)

}

Step 4: Finally, let us call the recommendProducts function on the model we built to show the top 10 recommendations.

println("\nTop 10 recommendations:")

val recommendations = model.recommendProducts(userID, 10)

for (recommendation <- recommendations) {

println( nameDict(recommendation.product.toInt) + " score " + recommendation.rating )

}}}

Your program should now look like the screenshot below. It is intuitively obvious that this will work with massive scale by tokenizing the userID variable to get recommendations for any user.

Task 5 is complete!

Task 6: Running the Model

Now that we have built the recommender system, let us finally run it.

Step 1: Run the model from your IDE. The model should run and provide the output as shown below.

As you can see from the screenshot above, the algorithm has recommended the top 10 movies based on our ratings. Though not all the movies are from comedy genre, most of them are.

Please note that you might not get the same recommendations when you run this program in your machine. You will also get different recommendations for every run.

Task 6 is complete!

REFERENCES

●  http://spark.apache.org/

●  https://spark.apache.org/mllib/

●  https://spark.apache.org/docs/2.2.0/ml-collaborative-filtering.html

●  https://dl.acm.org/doi/10.1109/MC.2009.263

Table of Contents

CHAPTER 1: INTRODUCTION TO BIG DATA & AI

Theory

An Overview of Big Data

Quick Introduction to Hadoop

Why Hadoop?

Quick Introduction to Hadoop Distributed File System

Block Placement in HDFS

HDFS Architecture

Introduction to MapReduce

Architecture of MapReduce

Processing Data with MapReduce

3V’s of Hadoop

Introduction to Spark

What is Spark?

Why Spark?

Components of Spark

Introduction to RDD

Architecture of Spark

Job Workflow in Spark

Introduction to SparkSQL

What is Spark SQL?

Why Spark SQL?

Spark SQL Architecture

What are Datasets?

Why Datasets?

Spark Data Storage

Various Spark Versions

Introduction to Artificial Intelligence

What is AI?

What is Machine Learning?

What is Deep Learning?

AIM

LAB EXERCISE 1: Installations

Task 1: Download and Install JDK

Task 2: Download and Install Scala

Task 3: Download and Install Spark

Task 4: Download and Install IntelliJ IDEA

Task 5: Configuring IntelliJ IDEA

SUMMARY

REFERENCES

CHAPTER 2: RECOMMENDER SYSTEMS USING APACHE SPARK

Theory

Basics of Machine Learning

Machine Learning Techniques

Getting Started with Recommender Systems

What is Recommender System?

Why Recommender Systems?

Recommender Systems Techniques

Content-Based Filtering

Collaborative Filtering

Spark MLlib Overview

Spark MLlib Components

Pipeline

Alternating Least Squares

AIM

LAB EXERCISE 2: Movie Recommender System using Apache Spark

Task 1: Download MovieLens dataset

Task 2: Creating a new package in IntelliJ IDEA

Task 3: Mapping Movie Ids with Movie Names

Task 4: Loading and Mapping User Data

Task 5: Training Recommendation Model

Task 6: Running the Model

REFERENCES