Apache Spark

INVENT THE FUTURE

ERNESTO LEE

APACHE SPARK

Copyright © 2021 by ERNESTO LEE

All rights reserved.

No part of this work may be reproduced, stored in a retrieval system, or transmitted in any form or by any means, electronic, mechanical, photocopying, or otherwise, without the prior written consent of the publisher. Short extracts may be used for review purposes.

Table of Contents

CHAPTER 1 INTRODUCTION TO APACHE SPARK

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

3Vs of Hadoop

Introduction to Spark

What is Spark?

Why Spark?

Components of Spark

Spark Data Storage

Various Spark Versions

LAB EXERCISE

SUMMARY

REFERENCES

CHAPTER 2: PROGRAMMING WITH SCALA

Theory

What is Scala?

Why Scala?

Data Types in Scala

Functions in Scala

Collections in Scala

Coding Scala

Conclusion

AIM

LAB EXERCISE 1: PROGRAMMING WITH SCALA

Task 1: Download and Install JDK

Task 2: Download and Install Scala

Task 3: Scala Basics

Task 4: Loops

Task 5: Functions

Task 6: Collections

LAB CHALLENGE

SUMMARY

REFERENCES

CHAPTER 3: HANDS ON SPARK

Theory

Introduction to RDD

Architecture of Spark

Job Workflow in Spark

AIM

LAB EXERCISE 2: HANDS ON SPARK

Task 1: Download and Install Spark

Task 2: Installing Spark on Multi-Node Cluster

Task 3: Creating RDDs from Spark-Shell

Task 4: Basic RDD operations

Task 5: Download and Install IntelliJ IDEA

Task 6: Configuring Intellij IDEA

SUMMARY

REFERENCES

CHAPTER 4: INTERNALS OF SPARK

Theory

Characteristics of RDD

RDD Operations

RDD Transformations

RDD Actions

Lineage Graph

Directed Acyclic Graph

AIM

LAB EXERCISE 3: SPARK PROGRAM

Task 1: Creating a new package in IntelliJ IDEA

Task 2: Spark Program – Loading Data

Task 3: Spark Program – Performing Operations

Task 4: Spark Program – Saving Data

Task 5: Spark Program – Lineage Graph

Task 6: Spark Web Interface

SUMMARY

REFERENCES

CHAPTER 5: RDD KEY-VALUE PAIRS & CACHING

Theory

Paired RDD

Paired RDD Transformations

Two Paired RDD Transformations

Paired RDD Actions

RDD Caching and Persistence

Persistence Storage Levels

AIM

LAB EXERCISE 4: PAIRED RDD – HANDS ON

Task 1: Creating a Tuple

Task 2: Creating a Paired RDD

Task 3: Performing Operations on Paired RDD

Task 4: Performing more Operations on Paired RDD

Task 5: Performing Joins on Paired RDDs

Task 6: Performing Actions on Paired RDDs

LAB CHALLENGE

SUMMARY

REFERENCES

CHAPTER 6: SHARED VARIABLES

Theory

What are Shared Variables?

Why Shared Variables?

Broadcast Variables

Optimizing Broadcast Variables

Accumulators

Points to remember when Accumulators are used

Scala Monadic Collections

Either Monadic Collection

Option Monadic Collection

Try Monadic Collection

AIM

LAB EXERCISE 5: SHARED VARIABLES – HANDS ON

Task 1: Using Accumulator method

Task 2: Implementing Record Parser

Task 3: Implementing Counters

Task 4: Implementing Accumulators V2

Task 5: Implementing Custom Accumulators V2

Task 6: Using Broadcast Variables

SUMMARY

REFERENCES

CHAPTER 7: SPARK SQL

Theory

Types of Data

What is Spark SQL?

Why Spark SQL?

Spark SQL Architecture

AIM

LAB EXERCISE 6: SPARK SQL – HANDS ON

Task 1: Creating Data Frame using Data Source API

Task 2: Creating DataFrame from an RDD

Task 3: Creating Data Frame using StructType

Task 4: Querying data using Spark SQL

Task 5: Joins using Spark SQL

Task 6: Operations using DataFrame API

SUMMARY

REFERENCES

CHAPTER 8: DATASETS

Theory

RDD vs. DataFrame

What are Datasets?

Why Datasets?

AIM

LAB EXERCISE 7: DATASETS & FUNCTIONS

Task 1: Creating Dataset using Data Source API

Task 2: Creating Dataset from an RDD

Task 3: Aggregate and Collection Functions

Aggregate Functions

Collection Functions

Task 4: Date/Time Functions

Task 5: Math and String Functions

Math Functions

String Functions

Task 6: Window Functions

SUMMARY

REFERENCES

CHAPTER 9: USER-DEFINED FUNCTIONS

Theory

Why User-Defined Functions?

Steps to implement User-Defined Function

UDAF Types

Function currying in Scala

Partially applied functions in Scala

AIM

LAB EXERCISE 8: USER DEFINED FUNCTIONS

Task 1: Defining Currying Functions

Task 2: Using partially applied functions

Task 3: Writing User Defined Function

Task 4: Writing Untyped UDAF

Task 5: Using Untyped UDAF

Task 6: Typed UDAF

SUMMARY

REFERENCES

CHAPTER 10: FILE FORMATS

Theory

DataSource API

Reading Data

Read Modes

Writing Data

Save Modes

Text Files

CSV Files

JSON

Parquet Files

ORC Files

RDD API

Text Files

Sequence Files

Hadoop Files

AIM

LAB EXERCISE 9: USING FILE FORMATS

Task 1: Text Files

RDD API

DataSource API

Task 2: CSV Files

Task 3: JSON Files

Task 4: Parquet Files

Task 5: ORC Files

Task 6: Hadoop and Sequence Files

Sequence Files

Hadoop Files

SUMMARY

REFERENCES

CHAPTER 11: SPARK CONFIGURATIONS & OPTIMIZATIONS

Theory

Spark Configurations

Spark Configuration Properties

Environment Variables

Logging

Performance Optimization

Using Datasets extensively

Avoiding UDF and UDAF

Data Serialization

Spark Memory Tuning

Level of Parallelism

Levels of Data Locality

Use Broadcast Variables

Filter Data as soon as possible

Logs

More Power

AIM

LAB EXERCISE 10: SPARK CONFIGURATIONS & OPTIMIZATIONS

Task 1: Spark Configuration File

Task 2: Using spark-submit Tool

Task 3: Environment Variables File

Task 4: Logging Properties File

Task 5: Checking Log Files

SUMMARY

REFERENCES

CHAPTER 1

INTRODUCTION TO

APACHE SPARK

Theory

This chapter provides a comprehensive introduction to Apache Spark, which is the center of focus throughout this book. Moreover, in the upcoming chapters, we will describe the Scala programming language to interact with Spark. But before we begin with Spark, let’s have a brief introduction of Hadoop and compare it with Spark.

An Overview of Big Data

Quick Introduction to Hadoop

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

The basic idea of Hadoop originated when Google released a white paper about Google File System (GFS) - a computing model developed by Google which was designed to provide an efficient and reliable access to data using large clusters of commodity hardware. Subsequently, this model was adopted by Doug Cutting and Mike Cafarella for their search engine called “Nutch”. Hadoop was then developed to support task and data distribution for the Nutch search engine project by Doug Cutting and Mike Cafarella. Well, what does the name Hadoop mean? There is no significance for this name and it is not an acronym either. Actually, Hadoop is the name that Doug Cutting’s son had given to his yellow stuffed elephant. Doug found this name is very unique, easy to remember and sometimes funny. Nonetheless, not only does Hadoop have such an interesting name with no significance, but its sub-projects also tend to have such names which are based on the names of animals like Pig and for similar reasons. Overall, these names are unique, not used anywhere else and are easy to remember.

Why Hadoop?

Companies today are realizing that there is a lot of information in the unstructured documents spread across the network. Basically, a lot of data is available in the form of spreadsheets, text files, e-mails, logs, PDFs, and other data formats that contain valuable information which can help discover new trends, designing new products, improving the existing products, knowing customers better and what not. Due to the emergence of many advanced technologies relying on the Big Data, many research studies have reported that the data explosion is increasing at an alarming rate beyond limits like never before and there are no signs of slowing down, at least in the near future. To deal with such data, we need a reliable and low cost tool to meaningfully process it. Therefore, Hadoop has been developed as a tool, which helps us to reliably process the Big Data, which is present in a variety of formats, in a very less time in a flexible and cost effective way.

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

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

● Fault-Tolerant: Hadoop helps prevent loss of data. All the data stored in the Hadoop Distributed File System (HDFS) is broken into blocks and stored with a default replication factor of 3. While processing this data, if a certain node goes off, the process does not stop but continues, as the data still exists in other nodes due to replication.

● Flexible: Hadoop does not require schema. Furthermore, it 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. It can efficiently work with a cluster of commodity machines by the process of 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 (e.g., a supercomputer). HDFS is a distributed large scale storage component and is highly scalable. Moreover, HDFS can accept node failures without any loss of data, and it is widely known for its reliability. Let us now check out why HDFS stands out of the crowd when it comes to the distributed file systems.

Reliable Data Storage

HDFS is very much reliable when it comes to data storage. The data stored in HDFS is replicated by a default replication factor of 3. It means that, even if a machine fails, the data is still available in two other machines.

Cost Effective

HDFS can be deployed on any custom-made clusters of commodity hardware and can save us a lot of bucks. Therefore, HDFS does not require any high-end or expensive hardware to function.

Big Datasets

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

Streaming Data Access

HDFS provides streaming access to data. It is best suited for batch processing of data and not suitable for interactive processing. Moreover, it is not designed for the 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 models for files. Appending the content to files at the end is supported, but files cannot be updated at any arbitrary point, and it is also not possible to have multiple writers. In addition, the 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, but the second replica is placed on a different rack to that of the first replica. Similarly, the third replica is placed on a random node on the same rack as the second replica. If the replication factor is more than 3, any random nodes in the cluster are selected to place the replicas. If a client running outside the cluster stores a file, a random node (which is not busy) is automatically picked to place the first replica. This way, if a node fails, the data is still available on other nodes of the cluster and if a rack fails, again, the data is still intact.

HDFS Architecture

HDFS is designed as 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 called:

● NameNode

● Secondary NameNode

The Slave Nodes in HDFS are called:

● Data Nodes

These nodes perform the core serving roles in HDFS architecture. Let us now look in detail at the roles of each Node to better comprehend them.

NameNode

NameNode is a 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. Moreover, it is also responsible for managing the File System namespace image. It holds the image in memory, representing how the File System looks like. It also maintains the meta data for all the blocks of files in the File System and tracks the replication value, so that it knows the locations of all the blocks stored on Data Nodes within the cluster. But the meta data is not stored on the disk and is recreated every time when it starts. NameNode stores all this information persistently on a local disk in the form of namespace image and edit log. The main limitation of the NameNode is that it is the single point of failure in the Hadoop cluster. Therefore, if the NameNode fails, the entire cluster fails.

Data Nodes

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

Secondary NameNode

Secondary NameNode, contrary to its name, is not exactly the Secondary NameNode. It is not a high availability solution and does not automatically take over the responsibilities of NameNode upon its failure. Primarily, its main role is to create a checkpoint and take the backup of NameNode periodically. More specifically, it is like a backup solution to the NameNode. The hardware specifications of the secondary NameNode should be similar to that of NameNode. Hence, in case of NameNode’s failure, the secondary NameNode can be manually configured to work as a primary NameNode. Therefore, it can be concluded that it is not a high availability solution.

Introduction to MapReduce

MapReduce is a programming model for processing large volumes of data stored in the distributed file systems, such as HDFS. This model is designed for low level programming and thus programs are written in low level languages such as Java, Python, and Ruby etc. Let us look at the architecture of MapReduce.

Architecture of MapReduce

MapReduce, similar to HDFS, is a master-slave architecture but instead of NameNode in HDFS, we have JobTracker and instead of Data Nodes in HDFS, we have Task Trackers in MapReduce model. Unlike NameNode and Data Nodes, the JobTracker and Task Trackers are not physical hardware components but Java programs (called MapReduce daemons) running on their own Jave Virtual MachineS (JVMs) inside the machines.

The detail description of MapReduce daemons is given as follows:

JobTracker

JobTracker is the MapReduce daemon which serves as a master to the Task Trackers. The role of a JobTracker is to receive job requests from the Hadoop clients and assign work to the task trackers on Data Nodes. It queries the location of data on Data Nodes and tries to assign task trackers on these Data Nodes where the data is present locally, in order to achieve data locality. If any node where the data is present locally, is busy, the JobTracker assigns tasks to another task tracker within the same rack.  Furthermore, if any task tracker fails for some reason, the JobTracker assigns the same task to another task tracker containing data, as the data is replicated across the cluster on multiple nodes.

JobTracker is the single point of failure similar to that of NameNode. If it fails, all the task trackers also fail and there will be no tasks running. Thus, it is wiser to spend more on the machine which runs JobTracker in order to decrease its chances of failure.

Task Trackers

Task Trackers are slaves to JobTracker, and they perform the actual work. Task Trackers accept tasks from JobTracker and execute these tasks. The Task Trackers regularly send their statuses or progress of the tasks to 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. In addition, the Task Trackers also send heartbeat messages to JobTracker about the free slots available with them for processing of any future tasks.

The failure of a Task Tracker is not as serious as that of a JobTracker as the JobTracker can always assign the failed task to an alternate Task Tracker.

Processing Data with MapReduce

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

● Map phase

● Reduce Phase

Both these phases have key-value pairs as input and output. The data types for the key-value pairs can be chosen by the developer. Moreover, the developer has also to specify a Map function and a Reducer function, which stipulate the necessary logic for processing the data in MapReduce. As their names suggest, the Map function serves as the logic to map tasks; whereas the Reduce function serves as the logic to reduce tasks.

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

● The input to map a task is in the form of a split. A split is a fixed chunk of data based on the input Format and should not be confused with an HDFS block. For clarity, the blocks belong to HDFS, whereas the splits belong to MapReduce. Block is the smallest size of data stored in HDFS, whereas a split is an input to map task. Generally, the optimal split size is equal to the block size. A single map task can process one input split at a time.

● The Map tasks process the splits using the Map function in parallel and produce an output. The output from the map tasks is not stored in HDFS but on the local disk because the map output is an intermediate result and there is no need to save it on HDFS with replications. The map output is only saved in the 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 is merged, sorted, and partitioned. Merging is the process where the data from all the map tasks is merged together. Similarly, sorting is the process where the map tasks output is sorted based on the key. Moreover, partitioning is a process where the data is divided based on keys, so that the values from all the keys should go to one reducer. Partitioning is useful when multiple reducers are utilized. Subsequently, the output of this Map task is fed to the reducer as an input.

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

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

Consider a file containing search terms for a website with each search term in a separate line as shown below:

Let us assume that each line is an input to different map tasks. Please note that, this assumption is for a better understanding of the concept and in reality each map task processes much more data than this. 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 will be sent to the mappers. The input splits of the file are as shown below:

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

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

When the intermediate data is processed through the shuffle and sort phase, the reducer starts producing the end result using the logic in the reducer function. After the shuffle and sort phase, the partitioner ensures that the key-value pairs with the same keys go to the same reducer if there are multiple reducer sets.

Subsequently, the reducer aggregates the input and produces the following output:

Then the output from the reducers is concatenated to produce a single file which contains the end result. This end result is as shown below:

In this manner, the end result shows the occurrences (i.e., frequency) of each word in a given file.

Finally, let us conclude the quick introduction on HDFS by looking at 3Vs of Hadoop, which summarize the overall capabilities of Hadoop.

3Vs of Hadoop

Hadoop can be better described with 3Vs, which are explained as follows:

● Volume: Hadoop is designed to process large volumes of data ranging from hundreds of Gigabytes to Petabytes. Currently, there are numerous datasets available in the scales of Petabyte, and Exabyte datasets are not a distant dream. Hadoop excels while processing such large volumes of data rather than small data.

● Velocity: Hadoop is designed to ingest data at higher speeds from multiple sources. For this purpose, Hadoop utilizes the distributed framework for parallel processing which in turn reduces the time taken to complete a job. Moreover, Hadoop brings computation to data rather than bringing data to computation; a paradigm which can easily reduce the bottlenecks related to network bandwidth.

● Variety: Hadoop can process data in a variety of forms including structured, semi-structured and unstructured data. Moreover, there are no restrictions on schema. Unlike relational database management systems, Hadoop has the schema on read capability. Therefore, no schema is required while writing to HDFS, and it can only be parsed at the read time.

The traditional computing models lack these features, but Hadoop excels by providing us with more power to explore our data.

Now that we have had a quick introduction to Hadoop, let us shift our focus on the main topic of our discussion: i.e., 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 with distributed processing of large datasets at high speed. Moreover, it supports most popular programming languages such as Java, Python, Scala and R etc. Spark is scalable, meaning it can be executed on a single desktop machine (or a laptop) as well as on a cluster of thousands of machines. In addition, it provides a comprehensive set of inbuilt libraries which can be accessed to perform data analysis over a large dataset. However, if our requirement does not get satisfied with the inbuilt libraries, we can also write one or explore countless external libraries from many 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 (it will be discussed later in this chapter). Furthermore, Spark is an in-memory processing framework, and it is considered as a successor of Apache Hadoop. Let us briefly discuss the advantages of Spark over Hadoop.

With the Hadoop ecosystem, we had a variety of frameworks for various data processing requirements. As a developer, we would use the MapReduce framework to analyze our data using any of the languages such as Java, C++, and Python etc., but a data warehouse engineer (an SQL expert), has to learn any of the aforementioned programming languages. To overcome this problem, a new framework which runs on the top of Hadoop called “Hive” had been introduced. But this also posed a problem for the ETL processing and therefore, “Pig” was introduced. Similarly, many other tools like “Giraph” and “Mahout” were introduced for graphs processing and Machine Learning, respectively. Another limitation of Hadoop is that it is only used for the batch processing and there was no way to process data in real-time. Therefore, a new framework called “Storm” was integrated with Hadoop to work with the real-time streaming data. Summarily, with so many different customized frameworks, the organizations found it extremely tough to maintain all the frameworks and track issues pertaining to them. Fortunately, all this would change with the advent of Apache Spark. As mentioned above, Spark is a unified platform which provides all these frameworks as one package with only four major components.

Now, what actually does the 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 finished, 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

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 required for any further processing, it has to be read again from the disk by a MapReduce application, followed by processing in memory and then it is written back to disk. This process continues as per the requirement, as observed in Figure 1(a). The processes of reading and writing data from and to the disk increase the IO latency, and therefore the overall job duration is increased. This situation has been optimized in Spark as observed in Figure 1(b).

1(b) Data Processing with Spark

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. Therefore, if the processed data is again required for any further processing, it 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 model, the processing speed has been increased up to 100 times faster as compared to MapReduce. The processed data from a Spark application can either be retained in memory or can be stored on the disk as per the requirement as shown in Figure 1(b).

The above-mentioned 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 switching to Spark for their data processing requirements.

Components of Spark

Now that we understand why Spark is being used, let us dive in more and learn of the components of Spark. Let us look at each of the major Spark’s components individually and comprehend them in detail. The following Figure 1(c) depicts the components of Spark.

1(c) Components of Spark

Let us look at 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 has all the basic functionalities for processing large datasets. Some of these functionalities include managing memory, scheduling jobs, fault tolerance, using in-memory computation, referring datasets stored in storage systems  etc. In addition, the Spark Core also includes a programming abstraction (i.e., API) called Resilient Distributed Datasets (also known as RDDs), which are responsible for partitioning data across nodes on a cluster. With the help of these RDDs, the data can be transformed, collected and reduced together. These RDD APIs can be referred to by using any of the programming languages such as Scala, Python, Java and R as depicted in 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 distributed over a cluster of nodes. The Spark SQL works well with structured and semi structured data. It can also work with various data sources, such as Apache Hive tables, Avro, JDBC, ORC, JSON and Parquet file formats. Moreover, Spark SQL also allows the developers to integrate the 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 deals with the processing of real time data, also known as the streaming data. The streaming data can be from a fleet of web servers, sensors, IOT devices or any other sources which have the ability to generate data. This enables Spark to ingest data as it is generated in real-time applications and perform data manipulation on that data. There are three major phases of Spark Streaming: They are called  Gathering, Processing and Data Storage. In addition, Spark Streaming is also fault tolerant and scalable. This book does not cover Spark Streaming.

Spark MLlib

Spark MLlib (short for Machine Learning libraries) provides a platform for creating Machine Learning models for huge datasets. It contains various well-known Machine Learning algorithms, such as  Regression, Clustering, Classification and Collaborative Filtering etc. Moreover, MLlib also contains lower level primitives such as generic gradient descent optimization algorithms. It also uses a linear algebra package called Breeze for numerical computing. This book does not cover Spark MLlib.

GraphX

GraphX deals with the processing of Graphs in a very efficient and distributed manner. GraphX extends the RDD APIs allowing 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 volumes of datasets by distributing them over a number of nodes and scaling the cluster as required. To efficiently perform this sort of tasks, a cluster manager is required; and Spark has its own cluster manager called  Standalone Scheduler. Moreover, Spark can also be deployed using Hadoop YARN, Apache Mesos or Kubernetes as cluster managers to schedule jobs and manage the resources of the cluster. We shall look into more about cluster managers in the upcoming chapters.

Spark Data Storage

Apache Spark supports major file systems such as HDFS, Amazon S3, Azure Blob etc. Moreover, it also supports the local file system for storing the data. However, using a distributed file system (such as HDFS) can leverage the power of Spark by distributing the datasets throughout the cluster. Furthermore, 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 instance, Apache Spark 1.2.1 means Spark has a Major version of 1 with Feature version as 2 and Maintenance version as 3. Please check the link in the references section to learn more about versioning in Spark.

Although 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 at the time of writing this book.

LAB EXERCISE

"There are no activities required for this lab."

SUMMARY

Apache Spark is an open source, fast and unified parallel large-scale data processing engine. It provides a single (unified) platform for various data processing tasks such as Big Data analytics, Streaming, Machine Learning (ML) and Graph processing. It has been reported that Spark is up to 100 times faster than MapReduce due to its In-Memory processing system and lazy execution of tasks.

Apache Spark has its own in-built cluster manager known as “Standalone Scheduler”, but it can also be deployed on various other popular managers, such as “Hadoop YARN”, “Apache Mesos” or “Kubernetes”.

REFERENCES

● http://spark.apache.org/

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

● http://hadoop.apache.org/

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

CHAPTER 2:

PROGRAMMING

WITH SCALA

Theory

Before we dig deeper into Spark, let us take a detour and look into the programming language called Scala, which we shall be using to write all our Spark code. This chapter covers most of the Scala basics to get started with Spark programmatically, which include the basics of Scala syntax, calling functions in Scala, using few data structures etc. It should be noted that Scala is the most popular language used to write Spark code, although Spark also supports many other programming languages such as Java, R, and Python etc.

What is Scala?

Scala is short for Scalable Language. It is called a scalable language due to its scalability as per the user requirement. Scala is a functional as well as Object-Oriented Open Source Programming language which runs on top of a Java Virtual Machine (JVM). Scala is, as mentioned above, the most preferred programming language for Spark, as Spark itself was built with Scala. It runs on top of the JVM; therefore, we can access Java libraries within the Scala code and utilize the functionalities of Java without being limited by Scala itself.

The compiler for the Scala code is called scalac, which compiles the Scala code to a byte code which can be understood by a JVM to process the code. This process is similar to how Java compiles its Java code by using the javac compiler to a byte code to process it on a JVM.

Why Scala?

Let us now briefly look at the advantages of Scala and know why it is the best fit with Spark.

Functional

Scala is a functional programming language. Every function in Scala is a value. The functions in the functional programming are implemented in such a way that there are no side effects. It means that the functions only perform what they are supposed to do. They take arguments, process the logic and return results. Scala supports higher order functions which takes a function within a function. The syntax for functions in Scala is lightweight, and it supports anonymous functions, function currying and nesting of functions. Moreover, Scala has in-built support for pattern matching model, as algebraic types are used in various functional programming languages. Scala’s case classes have classes which can be instantiated, so that they can automatically generate several methods. Scala also provides Singleton objects, which represent a class that has only one instance, to the group functions that are not the members of a class.

Object-Oriented

Scala is also a pure object-oriented programming language where every value is an object and every operation is a method call. An object in Scala is an instance of a class. Type and behavior of the object are described by classes and Traits. Traits are similar to that of interfaces in Java. However, traits can also have method implementations as well as fields.

Immutability

Variables in Scala can be declared as immutable using the ‘ val’ keyword. It means that once a variable is declared, it cannot be modified. Moreover, the collections in Scala are immutable by default, unlike Java, which are mutable by default.

Type Inference

Scala is smart enough to infer the data type. It means that we need not explicitly declare the data type of a variable while declaring or returning it in a function. Scala can determine the data type of the return type of a function based on the last expression within the function.

Pattern Matching

Pattern matching is a core feature of Scala. It is a process of checking for a specific pattern or a value in a given sequence of data. It is similar to that of  Switch in C and Java. In Scala, pattern matching is achieved by using the match method in a root class, and then a number of cases as an argument using the case keyword to perform a pattern match.

Interoperability

Scala supports interoperability with Java, as we can embed the Java code within the Scala code and run its code without any issues. Scala code is compiled to a bytecode to execute it on a Java Virtual Machine (JVM).

Data Types in Scala

Like any other programming language, Scala also has data types, and they are very similar to that of Java. The following are the data types in Scala:

Data Type

Size

Int

4 Bytes (32 bits)

Long

8 Bytes

Float

4 Bytes

Double

8 Bytes

Char

1 bit

Boolean

1 bit

String

Dynamic

Unit (Similar to void in Java)

--

Functions in Scala

Scala is a functional programming language, and therefore, functions play a crucial role. The syntax to define a function is as follows:

def function_name(parameters: type): return_type = {

// expressions

}

⮚ The function in Scala is declared by using the def keyword.

⮚ Next, the name of the function should be specified in the lower camel case.

⮚ Following the name of the function are the parameters or arguments and their type for the function separated by a colon. The parameters and their types should be enclosed in the parentheses. There can be multiple parameters and each of them should be separated by a comma.

⮚ The return type of the function can be specified after identifying the parameters. However, the return type is optional to specify, as Scala can infer the return type by looking at the last line of expression in the function body. If the function does not return anything, the keyword unit is used as the return type. This is similar to that of void in Java.

⮚ Finally, the return type is followed by an ‘equals’ operator and subsequently, the body of the function is defined, enclosed in the curly braces.

We shall look at the functions in the lab exercises.

Collections in Scala

Collections represent a group of variable numbers of heterogeneous data items or elements stored as a single object of a similar type. They can be manipulated according to the requirement. All the operations that we usually perform on data, such as searching, sorting, insertion, manipulations, and deletion, etc. can be performed by collection.

There are two types of collections in Scala, which are described as follows:

● Immutable collections

● Mutable collections

As the name suggests, the Immutable collections in Scala cannot be changed or modified once they are created, while the mutable collections can be modified after they are created.

Let us now look at a brief description of the most common collections.

List

List is the collection of elements with similar data types represented by a  linked list data structure. A List can also contain duplicate elements. Each element in the list can be accessed based on its position (index) starting from 0.

Set

Set is the collection of elements with similar data types but it does not contain any duplicate elements. Order of the elements inserted in a set is not maintained and therefore accessing the elements in a set based on their position (index) is not possible. Hashing techniques are used internally to store the values in the set collection.

Map

Map is a collection of key-value pairs, where keys are unique and their values can be retrieved based on the referenced keys. The values can also be accessed by the index of the key.

Array

Array is a collection of similar types of elements. An array in Scala is similar to that of Java. It is a mutable collection in Scala and its index starts with 0. Furthermore, Scala also supports multidimensional arrays.

Vector

A Vector is an improvised version of a List, which addresses the inefficiency in the random access on lists. Hence, vector shines where a list does not. A vector collection can have duplicates and the order of insertion of the elements is maintained.

Tuples

Tuple is a collection of items which can contain various types of elements. Unlike the other collections, tuples’ index starts with 1. Tuples are immutable and cannot be modified once they are set. They are frequently used in Spark.

Coding Scala

The Scala code can be written and executed in two ways, which are described as follows:

● Scala Shell (REPL)

● Scala IDE

Scala Shell is probably the easiest way to getting started with writing the Scala code. A Scala Shell is an interactive prompt utilized to write Scala expressions and programs. Moreover, the Scala Shell is sometimes also referred to as REPL which is short form for the read–eval–print loop.

Another way to write Scala code efficiently is to use a Scala plugin with any of the popular Integrated Development Environments (IDEs) such as IntelliJ, Eclipse, and NetBeans etc. In this book, we shall be using the IntelliJ IDE in all the upcoming chapters.

Conclusion

This concludes the theory part of Programming with Scala. Let us now proceed to the Scala labs to get our hands dirty and start programming with Scala.

Please note that this chapter has not been designed to make us a Scala expert. This is only designed as an introduction to Scala by covering the basic topics and to get us started with Apache Spark. Please refer to the References section to learn more about Scala.

AIM

The aim of the following lab exercises is to install Scala and perform various exercises by writing Scala code in order to get our hands on Scala.

The labs for this chapter include the following exercises:

● Downloading and Installing JDK

● Downloading and Installing Scala

● Getting started with Scala basics

● Learning the Loops concept

● Getting familiar with Functions

● Getting Familiar with Collections

Moreov,er we need the following packages to perform the lab exercise:

● Java Development Kit (JDK)

● Scala

LAB EXERCISE 1: PROGRAMMING WITH SCALA

1. Download and install JDK

2. Download and install Scala

3. Scala Basics

4. Loops

5. Functions

6. Collections

Task 1: Download and Install JDK

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

$ sudo apt-get update

This will update the package index. We might be asked to enter our password after we run this command.

Step 2: After we run the above command, we run the following command to actually download and install JDK.

$ sudo apt-get install default-jdk

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

The installation process may take some time depending on your internet connection and PC specifications. Please allow it to download and install completely. You should see the following message, after the installation is complete.

Step 3: Execute the following command to check if Java has been installed successfully. The terminal should show the Java version as depicted in the screenshot.

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

$ java –version

Task 1 is complete!

Task 2: Download and Install Scala

Now that we have successfully 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 command prompt will ask you to hit ‘Y’ after running the above command, as depicted 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 a Scala prompt will appear.

$ scala

This process completes the Scala installation. The scala prompt is the interactive shell where you can write and excecute the Scala code. This interactive shell is also known as REPL.

Step 4: We can now start writing the Scala code. Let us 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 we can see from the screenshot, the output appears below immediately as soon as we hit the Enter button.

Step 5: To quit the Scala REPL, we can use the following command:

scala> :q

We are now back to the terminal prompt.

Task 2 is complete!

Task 3: Scala Basics

Let us now familiarize ourselves with Scala and start writing some basic Scala code in order to get acquainted with the Scala syntax and basic programming constructs.

Step 1: Fire up your terminal and go into the Scala console by entering scala and hitting the Enter button. Once you see the Scala prompt, enter the following piece of code and hit the Enter Button.

scala> val name: String = “Learning Voyage”

The above line of code is utilized to declare an immutable variable named name of type String which has a value of Learning Voyage. The keyword val is used to declare an immutable variable. When val is utilized, the created variable can no longer be changed or modified. Generally, Scala encourages the immutability whenever possible. This can help us track our code easily and the values do not get modified accidentally while referring them programmatically. Unlike other languages, the name of the variable is declared first and then the data type is declared separated by a colon (:) in Scala.

We can now use the variable name and use it inside the println function as depicted in the figure below. This will print the value (String) associated with the variable (which is Learning Voyage in this case) to the console.

scala> println(name)

We can check if the declared variable is immutable by trying to append a new String to the name variable which we created previously as shown below.

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

scala> name = name + “ Inc”

As we can see from the screenshot above, it throws an error saying reassignment to val, which implies that we cannot reassign or modify an immutable variable.

Step 2: There might be scenarios where we would want to modify a variable. For that we can create a variable using var keyword instead of the val keyword as shown below.

scala> var newName: String = “Learning”

scala> newName = newName + “ Voyage”

scala> println(newName)

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

As we can see in the screenshot, we have concatenated a new string to the same variable and printed out the new string to the console using mutability. But it should only be used when there is an absolute requirement in a certain application.

Step 3: We can also apply a workaround to use the transformations on immutable objects to create new immutable objects and achieve the same result as we normally get by using the mutable objects.

scala> val name: String = “Learning”

scala> val newName: String = name + “ Voyage”

scala> println(newName)

As we can see from the screenshot above, we have applied the transformations on the immutable objects and achieved the same result as using the mutable objects.

Step 4: Let us now look into more data types as discussed in the Data Types in Scala section and observe how they can be created.

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

scala> val num: Int = 255

The above piece of code creates an immutable variable num of the type Integer with a value of 255. Similarly, we can create an immutable variable of all the other data types as shown below.

scala> val longNum: Long = 89416414

scala> val decimal: Double = 85.5545

scala> val decimalf: Float = 54.24f

scala> val letter: Char = ‘f’

Please note that there are only single quotes for the Char type, whereas there are double quotes for the String type.

scala> val lieDetector: Boolean = true

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

Please note that the data type name in Scala starts with an upper case letter unlike the other programming languages. However, we do not even have to specify the data type. Scala is smart enough to infer the type based on the value. We need not explicitly specify the data type while declaring a variable as shown below.

scala> val num = 256

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

As we can see from the screenshot above, Scala has automatically inferred the type of the variable as Int.

We can also do this for a String or any other data type as shown below.

scala> val name = “Learning Voyage”

scala> val decimal = 25.3545

Step 5: Let us now look at the various ways of printing to the console. Using all the variables we had created in the previous step, let us concatenate them all in one string using the ‘+’ symbol as shown below.

scala> println(“Printing to console using concatenation: ” + name + num + longNum + decimal + decimalf + letter + lieDetector)

This concatenation works but the output is not formatted correctly, because we have not used spaces to separate the variables. We can add a white space as a string after each variable, but it will become a lengthy process if we have too many variables.

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

Therefore, in order to overcome this issue, we can substitute the variables within a string using an $ prefix in the print statement before the double quotes as shown below. Each variable has a $ prefix.

scala> println(s“Printing to console using variable substitution: $name $num $longNum $decimal $decimalf $letter $lieDetector”)

Not only can we substitute the variables, but we can also replace expressions within the print statements enclosed in the curly braces.

scala> println(s“Four divided by two is ${4/2}”)

The Scala language also supports the printf statements similar to that of Java. All we have to do is to use an f prefix in the printf statement. An example of this process is given below:

scala> printf(f“Printing the value of a double with 2 decimal places $decimal%.2f”)

Task 3 is complete!

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

Task 4: Loops

Loops are an essential part of any programming language and it is no different with Scala. Let us now look at the loops concept and write some code to familiarize ourselves with loops:

Step 1: Let us start the loops concept with the if loop. Fire up the Scala console if you have not done already, and type in the following code.

scala> val numOfKids = 3

scala> if (numOfKids > 2) println (“They are Phoebe Buffay’s kids.”) else println (“Parent unknown!”)

As we can see from the screenshot, the console only prints out the statement which is true based on the condition.

We can also write the if loop in the REPL in multiple lines using the paste mode as shown below. From the Scala prompt, enter the following command and hit the Enter button.

scala> :paste

This takes us to the paste mode with a prompt to enter our code as shown in the screenshot.

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

We can now enter Scala code in multiple lines. Once we are done with our code, let us press Ctrl + D to come out of the paste mode and execute the code.

scala> :paste

//Entering paste mode (ctrl-D to finish)

val numOfKids = 3

if (numOfKids > 2) {

println(“They are Phoebe Buffay’s kids.”)

} else {

println(“Parent unknown!”)

}

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

The code is executed as soon as we have exited from the paste mode and the result is displayed as depicted in the screenshot above.

Step 2: Let us now look at the ‘for’ loops. Enter into the paste mode and execute the following code:

scala> :paste

//Entering paste mode (ctrl-D to finish)

for (i <- 1 to 5) {

val sum = i + i

println(sum)

}

The syntax for the ‘for’ loop is a bit unusual. We start the ‘for’ loop with the for keyword and assign a range of 1 to 5 both inclusive to the variable i. The <- symbol is the range operator in Scala. Basically, it means that the range of values between 1 and 5 are being assigned to the variable i  as a list of 1, 2, 3, 4, 5 and the loop is iterated through these values. Next, we declare a variable called sum and add each value 1 through 5 with itself. Finally, we print the sum of values by using the print line statement. The result is displayed with each value in a new line as shown in the screenshot below.

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

If we do not want the last iteration to be included, we can use the keyword until instead of to. For example,

scala> :paste

//Entering paste mode (ctrl-D to finish)

for ( i <- 1 until 5) {

val sum = i + i

println(sum)

}

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

We can also use an ‘if’ statement within the ‘for’ loop as shown below.

scala> :paste

//Entering paste mode (ctrl-D to finish)

val friends = List(“Chandler”, “Monica”, “Rachel”, “Ross”, “Joey”, “Phoebe”)

for(friend <- friends if friend == “Chandler”){

println(s“The king of sarcasm is $friend”)

}

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

In this example, we are looping through a list of collections called friends, with an if condition. We filter out all the items except for one element and substitute the variable in the print statement. Please note that we are using the double equals operators to compare the two strings.

Step 3: Let us now look at while and do while loops. The while construct is similar to that of any other programming language. However, in functional programming, the use of while loops is usually discouraged.

Enter into the paste mode and execute the following code.

scala> :paste

//Entering paste mode (ctrl-D to finish)

var friends = 0

val names = List(“Chandler”, “Monica”, “Rachel”, “Phoebe”, “Ross”, “Joey”)

println(“The names of friends are:”)

while (friends < 6){

println(s“${names(friends)}”)

friends += 1

}

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

In the code above, we have first declared an Integer variable with a value of 6 and then a list of names of type String. Next, we print out a header, so that the output makes sense and then write the While loop. The loop starts with a keyword while, and then the condition inside the parentheses. The condition we set here is to continue the loop until the value of friends is less than 6. Subsequently, we use the String interpolation to substitute the variables within the print statement. Please note that we have utilized the curly braces, as we have substituted a variable named friends as a value to the variable names, so that every time the loop runs, we are accessing each element of the list by its index starting from 0. Finally, we increment the variable friends with 1.

The while loop runs every time the condition is satisfied and only comes out of the loop if the condition is false.

Now, enter into the paste mode again, and execute the following code to perform a do while loop.

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

scala> :paste

//Entering paste mode (ctrl-D to finish)

var i = 0

do{

i += 1

println(i)

} while (i < 5)

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

The difference between while and do while loops is that any expression within the do clause runs at least once, irrespective of the condition in the while clause.

Step 4: Let us now look at the pattern matching, which is a core feature in Scala. Pattern matching is similar to that of switch statement in other languages.

Enter into the paste mode, and execute the following code:

scala> :paste

//Entering paste mode (ctrl-D to finish)

val job = “Transponster”

job match {

case “Masseuse” => println(“That’s Phoebe”)

case “Chef” => println(“That’s Monica”)

case “Executive” => println(“That’s Rachel”)

case “Transponster” => println(“That’s Chandler”)

case “Actor” => println (“That’s Joey”)

case “Paleontologist” => println(“That’s Ross”)

case _ => println(“Unknown job role”)

}

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

In the code above, we have created a new variable called job which has a value of Transponster. Subsequently, we use the match keyword to match the job variable with a list of cases. As we can see from the screenshot, the value of job is correctly matched and the output is printed. The last case statement with an underscore (_) is a wild card operator. It is used so that, if none of the cases match, the default case is executed in the loop. Moreover, please note that there are no break clauses in Scala, similar to that of Java. Scala has an in-built fall through the mechanisms and hence there are no break statements required.

Task 4 is complete!

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

Task 5: Functions

Task 1: Let us first create a function which does not have any parameters or a return type. Enter into the paste mode in the Scala console and execute the following code:

scala> :paste

//Entering paste mode (ctrl-D to finish)

def hello = {

println(“Hello there!”)

}

Now, exit out of the paste mode and simply call this function by its name.

hello()

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

As we can see from the screenshot above, Scala has automatically inferred the return type as unit which implies that there is no return type. As described earlier, Unit is similar to that of Void in Java.

Step 2: Let us now create a function which takes parameters and has a return type.

scala> :paste

//Entering paste mode (ctrl-D to finish)

def married(name: String, times: Int): String = {

return name + “ has married ” + times + “ times”

}

Now, exit out of the paste mode and simply call this function by its name.

married(“Ross”, 3)

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

Please note that the return type (which is String in this case) and the keyword return are optional. Scala can determine the return type based on the last expression in the function body as shown below.

Step 3: Let us now look at Higher Order Functions, which are the special types of functions which take other functions as their parameters. Let us understand them with the following example:

In the Scala console, enter the following code:

scala> def squared (num: Int) : Int = {

num * num

}

We can call the function square to see if it works. For example,

Scala> squared(5)

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

Now, let us pass this function as a parameter to another function.

scala> def highSquared(num: Int, func: Int => Int): Int = {

func(num)

}

Here, we have defined a function named highSquared which takes two parameters and returns an Int. One of these parameters is an integer named num and the other one is a function named func which takes a parameter of type Int and returns an Int. In the function body, the function func takes the value of num and returns its value. Let us invoke this function.

scala> val result = highSquared(4, squared)

scala> println(result)

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

Step 4: We can also use the function literals in the parameters instead of using the function name as shown below.

scala> highSquared(6, x => x * x)

In the example above, we have used a function literal instead of referring to the name of another function. Basically, it does the same thing as we performed in the previous step, but in a different way. This methodology is often used in Spark and it is very imperative that we must understand what is going on here.

Task 5 is complete!

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

Task 6: Collections

Let us now look at a few collections in Scala.

Step 1: Let us create a list and apply various transformations or operations over it.

scala> val shows: List[String] = List(“F.R.I.E.N.D.S”, “The Big Bang Theory”, “Game of Thrones”, “Breaking Bad”, “The Mentalist”)

Mentioning the type of list is optional in Scala, as it can infer the type automatically. We can simply type “val shows = ” and continue with the list.

Let us now print the list.

scala> println(s“Some of the popular TV shows are: $shows”)

Furthermore, we can access the individual items in the list using their indices.

scala>println(s“The TV show at position 0 is ${shows(0)}”)

scala>println(s“The TV show at position 1 is ${shows(1)}”)

scala>println(s“The TV show at position 4 is ${shows(4)}”)

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

Let us now learn how we can access the first and the rest of the elements by using head and tail.

scala>println(shows.head)

scala>println(shows.tail)

Let us now use foreach to print each element in the list.

scala> shows.foreach(println)

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

Let us now utilize the map function to convert all the show titles to the lower case.

scala> shows.map(show => show.toLowerCase)

These are only a few transformations, we can apply over a list. However, there are so many more transformations and operations we can apply on. From the Scala console, simply type the list of the name and the dot operator as shown below and press tab key on your keyboard. The console should display a long list of transformations we can apply on our list.

scala> shows.<press tab key>

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

If we are not sure what a function does, we can simply type its name after the dot operator and press the tab button twice. The console will automatically show us what the function expects us to pass in. For example,

scala> shows.reduce<press tab key twice>

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

We can play around these different transformations, and search them over the internet, if we do not understand any of them.

Step 2: Let us now look at the map collection. Let us first create a map of elements.

scala> val couples = Map(“Chandler” -> “Monica”, “Ross” -> “Rachel”, “Phoebe” -> “Mike”)

Now that we have a map collection, let us try to access the value by the key.

scala> println(couples(“Chandler”)

As we can observe from the screenshot above, we were able to access the value based on its key.

But if we try to access a value for a non-existing key, an exception is thrown as depicted in the screenshot.

scala> println(couples(“Joey”)

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

To overcome this problem, we can use the getOrElse method and specify a default value when the key does not exist.

scala> val unknown = util.Try(couples(“Joey”)) getOrElse “Not Known”

scala> println(unknown)

We can play around with Map, as we did with the lists and explore all the transformations and operations, we can perform on the Map objects.

Step 3: Let us now create a tuple and see how we can access its elements.

scala> val showInfo = (1994, “Friends”, 8.8, 2011, “Game Of Thrones”, 9.4, 2010, “Sherlock”, 9.1)

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

As we can see from the screenshot above, a tuple can contain various types of elements. Moreover, we need not explicitly specify the data types for the elements, as Scala can infer the data types automatically.

Let us now access the elements of the tuple based on its index. Please remember that the index of a tuple starts with 1 and NOT with 0.

scala> println(showInfo._1)

scala> println(showInfo._5)

We can also access the elements of a tuple and print them out to the console as shown below.

scala> println(s“${showInfo._5} is the highest rated show with “${showInfo._6} rating.”)

JDK D/I & Ins  >> Scala D/I & Ins  >>  Scala Basics

Loops

Functions

Collections

As always, we can play around with tuples and practice as much as possible.

Task 6 is complete!

Note: This chapter does not cover everything from Scala. There are numerous other topics. Therefore, please check the references section to learn more about Scala.

LAB CHALLENGE

● Write Scala code to perform basic mathematical operations, such as additions, subtractions, multiplication and division.

● Write a Scala program to print all the data types using various print techniques.

● Write a Scala program to list all the natural numbers below 10 which are the multiples of 3 or 5. Consequently, we will get 3, 5, 6 and 9. The sum of these multiples is 23.

● Write a Scala program to list the Fibonacci sequence, whose values do not exceed two million.

SUMMARY

Scala is a Functional as well as Object Oriented Open Source Programming language which runs on top of the Java Virtual Machine (JVM). In this chapter, we have learned why we need Scala and its various features, such as Functional Programming and collections etc.

In the labs, we have learned to install JDK and Scala to our machines and learned the Scala basics, loops, functions and collections.

REFERENCES

● https://www.scala-lang.org/

● http://allaboutscala.com/

CHAPTER 3:

HANDS ON SPARK

Theory

It is now time to get our hands on Spark. In the previous chapters, we have covered the Introduction to Spark and focused on the basics of Scala programming Language. In this chapter, let us take a deep dive into the internals of Spark and learn how Spark works. Later in the labs, we will be installing Spark on our machines and run the Spark code.

Introduction to RDD

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

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

Distributed: A basic property of 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, and database file via JDBC etc.

An RDD can be created in two ways. We can use the parallelize keyword on an existing collection or use the process of 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, whereas Actions are the operations where a result is returned after computing an RDD.

We shall look at RDD in more detail in the next chapter. Let us now look at the architecture of Spark.

Architecture of Spark

As we know from Chapter 1, Spark is a unified parallel large-scale data processing engine. To achieve parallelism for processing large volumes of data efficiently and with fault tolerance, Spark framework had been developed. But how does the Spark system work?

Let us understand the basics of data processing, so that we would know the importance of processing data in parallel manner. Consider processing data on your Personal Computer (PC). You have a text file of size ranging from a few MBs to a few hundred MBs. You can easily load this file to your PC and do various operations such as searching, sorting and filtering without any issue. Now, consider a plethora of text files each ranging from a few hundreds of MBs to a few hundreds of GBs, or even a few hundreds of TBs. Let alone processing such a big data, it becomes even harder to store those files in the first place. As a result, this becomes a bottleneck for your PC. You now have only two options. You can either increase the specifications of your PC, or process a percentage of files on multiple computers. The first option is not feasible as you cannot keep on upgrading hardware to a single computer. Therefore, the second option looks more feasible. But how will you coordinate the processing on each PC to be efficient and accurate? How do 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 certain computer in your network stops working due to a hardware problem?

Apache Spark is the answer for all these questions. Spark takes care of managing the group of computers (nodes) called clusters and coordinating the tasks on these nodes to process data.  Spark utilizes a cluster manager to manage a cluster. As witnessed in Chapter 1, there are various cluster managers such as Spark’s standalone cluster manager, YARN, Mesos and Kubernetes etc. The cluster manager takes care of managing and coordinating the Spark applications.

As we have already learned, Spark is 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 known as the Spark Session, which provides access to all the underlying hardware for the Spark job. Therefore, we can think of Spark Session as a gateway to access all the Spark functionalities similar to a database connection.

3(a) Spark Architecture

Prior to Spark 2.0, SparkContext was the entry point or a gateway for any Spark application. RDD was one of the main APIs of Spark which could be accessed through SparkContext. Similarly, to access other APIs other contexts were required, and the developers had to create various contexts, such as SQL Context for Spark SQL, Hive Context for Hive queries and Streaming Context for Spark Streaming. Thus, 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 whereas, the cluster manager manages the resources on the cluster. As we discussed earlier, the cluster manager can be Spark's standalone cluster manager, such as YARN, Mesos or Kubernetes etc.

The slaves are the worker nodes which are known as Executors in the Spark. As the name suggests, executors are the workhorses of Spark and are responsible to perform 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, we can add more executors as our data increases.

The Spark session is responsible to break the job and distribute the work to the executors, whereas 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 job workflow in Spark.

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

● In the next 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 execute the job are performed at this step.

● Subsequently, the driver contacts the cluster manager to allocate resources to execute 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 their respective heartbeats to the driver while executing the tasks.

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

● Once all the executors complete their tasks, their respective results are sent back to the driver through the cluster manager.

We have found some new terms in this chapter, such as the Directed Acyclic Graph (DAG) and the RDD Lineage Graph. Please, do not worry about them, as we will be looking at them in detail in the next chapter and they will make more sense then.

Let us now move to the labs and install Spark and Intellij IDE to write Spark code.

AIM

The primary aim of the following lab exercises is to install Spark and Intellij IDE. Moreover, we shall also perform various exercises by writing Spark code, so that we can get our hands on Spark.

The labs for this chapter include the following exercises:

● Downloading and Installing Spark

● Installing Spark on Multi-Node Cluster

● Creating RDDs from Spark-Shell

● Learning basic RDD Operations

● Downloading and Installing IntelliJ IDEA

● Configuring the IntelliJ IDEA

We need the following packages to perform the lab exercise:

● Java Development Kit (JDK)

● Scala

● Spark

LAB EXERCISE 2: HANDS ON SPARK

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

1. Download and install Spark

2. Installing Spark on Multi-Node Cluster

3. Creating RDDs from Spark-Shell

4. Basic RDD Operations

5. Download and Install IntelliJ IDEA

6. Configuring IntelliJ IDEA

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

Task 1: Download and Install Spark

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

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

$ java –version

We can see the version displayed in the terminal as shown in the screenshot above. If we do not see the Java version displayed, our machine does not have Java installed. Please install JDK before continuing to the next step.

Step 2: Let us install Spark in the 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 it, click the link as shown in the screenshot below to download Spark.

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

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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

Step 3: The downloaded file will be saved to the Downloads directory by default. Execute the following command to change the directory to Downloads folder.

$ cd Downloads

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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

$ ls

Now that we are sure that we 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 begin to untar to /usr/share directory as depicted in the screenshot above. We can verify the same by executing the following command below:

$ cd /usr/share

$ ls sp*

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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

Let us make a softlink to the Spark directory, so that we do not have to refer to Spark with the complete name as above. Moreover, this is also useful for future updates. Execute the following command:

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

Now we run the following command again to check if we were able to create the softlink successfully.

Step 4: Let us now set up the environment variables for Spark by executing the following command:

$ sudo vi ~/.bashrc

The file should open as shown in the screenshot below.

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

Now press the ‘’' 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

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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

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

$ source ~/.bashrc

Step 5: Let us now test the Spark installation by accessing the Spark Shell by executing the following command.

$ spark-shell

Once we run the above command, a screen similar to the screenshot below shows the confirmation of the successful installation of Spark.

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

Task 1 is complete!

Task 2: Installing Spark on Multi-Node Cluster

In the previous task, we have installed Spark on a single node. Let us now look at the steps for installing Spark on a Multi-Node cluster. Thus, in this task, we will install Spark on multiple nodes. For this purpose, we can use a cloud-based platform such as Amazon AWS. In this case, we shall be setting up a cluster of 3 nodes: 1 Master and 2 Slaves.

Step 1: Let us first install Spark on the Master node. Edit the host file by executing the following command, and adding the IP address of master and slave nodes.

$ sudo vi /etc/hosts

We will be shown an editor as shown below.

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

Append the IP address of the master and the slave nodes to the hosts file as follows:

<Master IP>      master

<Slave01 IP>      Slave01

<Slave02 IP>      Slave02

Please replace the <Master IP>, <Slave-1 IP> and <Slave-2 IP> with their respective IP addresses as shown below.

Save and close the file in order to return to the console.

Step 2: Install JDK and Scala by following the steps in the previous exercise. Please note that we can only proceed to the next step after installing JDK and Scala.

Step 3: Let us now configure SSH, so that the nodes can connect with each other. Execute the following command to Install Open SSH Server-Client.

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

$ sudo apt-get install openssh-server openssh-client

Generate the key pairs by executing the command below:

$ ssh-keygen -t rsa -P ""

Next, set up the passwordless SSH by copying the public key .ssh/id_rsa.pub from Master to the .ssh/authorized_keys of Slave nodes.

Once copied, we should be able to SSH to the slave nodes without being prompted for password.

Step 4: In this step, we shall download and install Spark. Let us follow the steps in the previous task till step 4 to install Spark before continuing with this step. Once we are able to set up the environment variables, let us execute the following command to edit the spark-env.sh file. The file is present in the $SPARK_HOME/conf/ directory.

$ cp $SPARK_HOME/conf/spark-env.sh.template $SPARK_HOME/conf/spark-env.sh

$ sudo vi $SPARK_HOME/conf/spark-env.sh

Once we are in the editor, let us append the following parameters to the file.

export JAVA_HOME= /usr/lib/jvm/java-11-openjdk-amd64/bin

SPARK_WORKER_CORES=16

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

Please note that our Java path might be different. Please run the following command to know the Java directory path.

$ update-alternatives --list java

Step 5: Finally, create a file named slaves in the $SPARK_HOME/conf/ directory.

$ sudo vi $SPARK_HOME/conf/slaves

Now, add the following entries in the slaves file.

Slave01

Slave02

This completes the Spark installation on the Master node. We now have to install it on the Slave nodes.

Step 6: Please repeat the steps 1 and 2 of this task on each Slave node. Once we are done with the steps 1 and 2, let us create a tar file of Spark installation by executing the following commands.

$ tar -zcvf spark.tar.gz /usr/share/spark

The tar file is saved to the home directory when we run the command above. Copy the tar file from the Master node to all the Slaves on the cluster.

After we have copied the tar file to slaves, untar the Spark tar file on all the slaves of the cluster by utilizing the following command:

$ tar –xvf /usr/share/spark.tar.gz

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

With this, we have successfully installed and configured Spark. We are now ready to start executing the Spark in distributed mode.

Step 7: We can now start the Spark daemons by running the following command from the Master node:

$ /usr/share/spark/sbin/start-all.sh

We can also stop the Spark daemons by running the following command from the Master node:

$ /usr/share/spark/sbin/stop-all.sh

We may also choose to start and stop the Master and Slave nodes by running the following commands:

$ /usr/share/spark/sbin/start-master.sh

$ /usr/share/spark/sbin/start-slaves.sh

$ /usr/share/spark/sbin/stop-master.sh

$ /usr/share/spark/sbin/stop-slaves.sh

This task was created for the information purposes only and we do not have to use the multi node cluster for the rest of the exercises.

Task 2 is complete!

Task 3: Creating RDDs from Spark-Shell

Now that we have Spark installed in our machines, let us begin coding with Spark. As a first step in our journey with Spark coding, let us look at the different ways to create an RDD.

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

Step 1: Let us create an RDD by using the parallelize keyword. Fire up the spark-shell from the terminal, and create a list as shown below.

$ spark-shell

scala> val friends = List(“Chandler”, “Rachel”, “Phoebe”, “Joey”, “Ross”)

Now, let us use the parallelize keyword and create an RDD for the list, we have created above.

scala> val friendsRDD = sc.parallelize(friends)

Now we have created our first RDD successfully. Please note that we are using the sc which is the object of SparkContext. The sc object is automatically created when we launch the Spark Shell as depicted in the screenshot below. This is the reason we are able to access sc and use SparkContext.

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

Step 2: Let us now create an RDD using a file from the file system. We shall be utilizing the textFile API to create an RDD from the file system. First, download the file ratings.csv from the URL below and save it to the home/chapter_3 folder. (Please create a folder named chapter_3 in the home folder.)

ratings.csv - http://bit.ly/2L8IEBS

Each line of this file represents the rating of one movie by a user, and has the following format: userId, movieId, rating, timestamp.

scala> val ratings = sc.textFile(“chapter_3/ratings.csv”)

Step 3: We can now create a new RDD from the existing RDD. For instance, let us count the number of ratings in the ratings RDD we had created in the previous step.

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

scala> val count_ratings = ratings.count

As we can see from the screenshot above, the count of all the records (ratings) present in the RDD has been returned as a new RDD called count_ratings.

Step 4: Moreover, we can also create an RDD from data present in the Hadoop Distributed File System (HDFS) using the same textFile API. But instead of a local path, we have to provide a HDFS path.

scala> val ratings = sc.textFile(“hdfs://dev_server:9000/file.txt”)

Similarly, the RDD is created using the data from HDFS and then we can continue to apply the transformations and actions.

Task 3 is complete!

Task 4: Basic RDD operations

Step 1: Let us start learning the basic RDD operations in Spark by creating an RDD from a collection as we did in the previous task.

scala> val letters = List(f, a, g, f, c, a, b, n, d, b)

scala> val lettersRDD = sc.parallelize(letters)

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

Step 2: We can access the first element of the RDD using the first method as shown below.

scala> lettersRDD.first

As we can see from the screenshot above, the first element in the RDD has been returned.

We can also use the take(n) method to read the n elements from our RDD, where n is the number of elements starting from the 1st element we want to read.

scala> lettersRDD.take(4)

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

But if we want to view all the elements in the RDD, we have to use the collect method as depicted in the screenshot below. Please note that using collect on a large dataset is not recommended, as collect will bring all the data of an RDD to the driver program and load it in its memory.

scala> lettersRDD.collect

Step 3: Let us now use a filter function using the ‘contains’ method and filter out an RDD which satisfies the filter criteria. Let us create a new list as shown below and then filter out a string:

scala> val friends = (“Monica”, “Chandler”, “Ross”, “Phoebe”, “Rachel”, “Joey”)

scala> val friendsRDD = sc.parallelize(friends)

scala> val chandler = friendsRDD.filter(name=> name.contains(“Chandler”))

scala> chandler.collect

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

The filter function we have used above is a Higher Order Function which takes another function as a parameter and returns an RDD of type String. The name => name.contains(“chandler”) is similar to a function in Scala as depicted below.

scala> def find(name: List[String]): Boolean = {

name.contains(“Chandler”)

}

Let us call the function with the parameter friends which is a List of type String.

scala> find(friends)

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

The code is given below.

friendsRDD.filter(name=> name.contains(“Chandler”))

is the same as the following code,

def find(name: List[String]): Boolean = {

name.contains(“Chandler”)

}

friendsRDD.filter(find)

Step 4: Let us now use a map function on the friendsRDD and output a tuple with the first character in each element and the name itself.

scala> val pairs = friendsRDD.map(name => (name.charAt(0), name))

With this process, the pairs RDD gets created. Now let us use the foreach keyword to print each element of the pairs RDD.

scala> pairs.foreach(println)

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

As we can observe from the screenshot above, we have used the map function to create a tuple with the first character of the name of each element and name itself in the friendsRDD. The first character is obtained by calling the function called charAt which takes the number to access the position of a character from a String.

If we think this is a bit complicated to understand, let us look at another example with a simple map function. Let us create a List of few numbers and create an RDD from that list as shown below:

scala> val num = List(1, 2, 3, 4)

scala> val numRDD = sc.parallelize(num)

Now let us write a map function which takes the numRDD and gives a squaredRDD as shown below:

scala> val squaredRDD = numRDD.map(x => x * x)

scala> squaredRDD.foreach(println)

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

numRDD.map(x => x * x)

is same as

def square(x: Int): Int = {

x * x

}

numRDD.map(square)

Step 5: For the numRDD we had created in the previous step, let us utilize the reduce function to add all the numbers.

scala> val sumRDD = numRDD.reduce((a, b) => (a + b))

Similarly, we can also utilize the reduce function to multiply all the numbers in numRDD.

scala> val mulRDD = numRDD.reduce((a, b) => (a * b))

These are a few basic RDD operations. We shall learn more operations in detail in our next chapter which is entirely dedicated to RDDs.

Task 4 is complete!

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

Task 5: Download and Install IntelliJ IDEA

Till now, we have been writing our code in the Spark Shell, which is great for the development of applications with small datasets. But in the real time, we may be utilizing an Integrated Development Environment (also known as IDE) for developing our Software applications.

We shall be using IntelliJ IDEA to write and execute the Spark-Scala code throughout the lab exercises. In this task, we shall download and install the IntelliJ IDEA and then install the Scala plugin, so that we can write the Scala code.

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

http://bit.ly/2V1HFYO

The file should begin to automatically download to the Downloads folder. It might take some time depending upon your internet connection.

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

Step 2: Once the download is complete, open the terminal and execute 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 our versions of IntelliJ might be different. Please replace the version correctly.

Step 3: Now let us execute the following command to install IntelliJ:

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

Please note that our path or the version might be different.

We should now see a prompt asking to import settings. Hence, simply click “OK”.

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

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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.

We should now see the welcome screen as shown below:

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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

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

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

This action should begin the download process. Once it is downloaded, we will be asked to restart the IDE. Please click on the “Restart IDE” button. Click “Restart” in the confirmation pop-up.

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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

Task 5 is complete!

Task 6: Configuring Intellij IDEA

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

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

Step 2: We will then be taken to the “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, we will be prompted to enter the project name. Enter the project name as “Spark”. Let us ensure that the JDK, SBT and Scala versions are selected automatically as depicted in the screenshot. Moreover, check the “Sources” checkbox for both SBT and Scala if they are not already checked. Finally, click on the Finish button.

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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

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

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

http://bit.ly/2UQvnNU

Click on the Spark Project Core link as depicted 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 the 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 them into the build.sbt file.

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

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

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

Sparks/I & Ins  >> Sparks D/I & Ins  >>  Creating RDDs

RDD op

IntelliJ D/I Ins

Configure IntelliJ

We are now ready to write our first Spark program!

Task 6 is complete!

SUMMARY

Resilient Distributed Dataset (also known as RDD) is the basic data structure of Spark, which is an immutable and fault-tolerant collection of elements which can be computed in parallel over a cluster of machines.

Spark has a master-slave architecture, and it consists of a Driver Program as the Master and executors as Slaves. A cluster manager is used to manage the resources across the cluster.

In the labs, we have installed Spark to our machines and learned the basic RDD operations. Moreover, we have also installed and configured IntelliJ IDEA as IDE for our Spark code.

REFERENCES

● https://spark.apache.org/

CHAPTER 4:

INTERNALS OF SPARK

Theory

In the previous chapter, we described a brief introduction to the Resilient Distributed Datasets (RDDs). Let us now delve deeper and learn more about RDDs, which are the fundamental data structures of Spark as they provide the core abstraction. Moreover, let us look in detail what goes on under the hood when we submit a Spark job.

Let us recall what we have learned in the previous chapter. We discussed that an RDD is a collection of records (elements) which are distributed across the memory of nodes of a cluster and is tolerant to data loss. Moreover, RDDs are immutable and cannot be modified once they are declared. In addition, RDDs are evaluated by Spark lazily. Below are the characteristics of RDDs, which have been explained in detail.

Characteristics of RDD

Immutability

As discussed earlier, RDDs are immutable and cannot be changed once declared. However, an RDD can be transformed to a new RDD by applying operations.

In-Memory

RDDs are computed In-Memory where the data is retained within the memory instead of spilling the data to disk. The methods  cache() and persist() are utilized to retain the data in memory. Please check “Why Spark” in Chapter 1 for the detailed information.

Lazy

RDDs are evaluated lazily, which means that when a transformation is applied to an RDD, it is not evaluated immediately but is only evaluated when an action is applied to the transformations.

Tolerant

RDDs are tolerant to the data loss, and any data loss can be rebuilt using the lineage graph.

Partitions

RDDs are broken down into logical chunks called partitions and are distributed across the nodes in a cluster. Basically, the partitions help achieve parallelism.

RDD Operations

There are two types of operations that can be applied on RDDs. They are known as the Transformations and Actions. Let us now look at them in detail.

RDD Transformations

Transformations are the operations where a new RDD is created from an existing RDD by applying a function. The existing RDD is not modified, as it is immutable. Therefore, a transformation is applied on an existing RDD and a new RDD is created. Transformations are lazy, as a transformation applied to an RDD is not evaluated until an action is called. Each transformed RDD indicates a pointer to a parent RDD. All these pointers along with the RDD dependencies are logged in a graph called the Lineage Graph, which creates a Logical Execution Plan. Whenever we perform a transformation or a series of transformations, they are not applied on the data right away but they are logged in the Lineage Graph as mentioned above. We shall look at Lineage Graph in the next section. The examples of transformation functions are map, flatMap, filter, reduceByKey, join, union, and sample etc.

There are two types of transformations, which can be applied on an RDD. They are described as follows:

Narrow Transformations: The transformations where the data required to perform a transformation exists in a single partition and no data is shuffled from other partitions are called Narrow Transformations. The examples of Narrow Transformations are map and filter etc.

Wide Transformations: The transformations where the data required to perform a transformation has to be shuffled across various partitions are known as Wide Transformations.  The examples of Wide Transformations are groupByKey and reduceByKey etc.

RDD Actions

Actions are the operations where the actual processing of data happens. They are either used to display or store the final result to a location. The transformed RDDs are evaluated once an action is called, triggering the execution using a lineage graph. The examples of actions include collect, count, countByValue, take, and top etc.

The transformations and actions help improve efficiency by lazy evaluation.

Lineage Graph

In the previous section, we learned that the transformations are applied to RDDs in order to create new RDDs, which are not evaluated immediately, but lazily. When a transformation function is applied to an existing RDD, a new RDD is created with a pointer to the parent RDD. These pointers are logged to a Lineage Graph along with the dependencies of all the RDDs. Let us consider an example to understand this well.

● A new RDD is created by using the textFile API. Let us call this RDD1.

● A flatmap function is applied on RDD1, to create a new RDD called RDD2.

● A filter function is applied on RDD2, creating a new RDD called RDD3.

● Finally, the result (RDD3) is stored by using the saveAsTextFile API, which is an action.

The first three operations in the above example are the transformations, which are not evaluated immediately, but lazily and the last operation is an action. The action triggers the evaluation of RDDs and the result is stored to the disk. The Lineage Graph for the operations above should look as shown below.

4(a) Lineage Graph

The Lineage Graph helps in recovering from the data loss in partitions of an RDD, as it has to simply perform the transformations again for the lost partition of an RDD. Consider the example above. If RDD3 is lost; Spark can simply apply the transformation on RDD2 and recover it. This feature of Spark helps in being fault tolerant without having to replicate data on multiple nodes as HDFS. This is how Spark is tolerant to the data loss.

Directed Acyclic Graph

Directed Acyclic Graph (also known as DAG) is a graph with vertices and edges connecting each other in a non-circular manner. In simple words, DAG is a graph with dots and lines pointing to other dots in one direction where the lines do not form circles to go back to the same dots. An example of a DAG is a Tree Graph. Look at the Fig 4(a) below to better understand a DAG.

4(b) Directed Acyclic Graph

As we can observe from Fig 4(a), each node (dots or circles with a number) points to another node in one direction (directed) and none of the nodes have pointers back to the same node (acyclic).

But what does DAG have to do in the context of Spark? Well, Spark uses DAG to optimize and efficiently run the Spark jobs. As we have learned, DAG is a graph with vertices and edges connecting each other in a non-circular manner. In Spark, RDDs are the vertices and the operations, which we apply on the RDDs, are the edges. As we have witnessed in the Lineage Graph above, each operation is directed to another. In other words, each vertex directs an edge to another vertex.

Nevertheless, how is a DAG different from Lineage Graph? A Lineage Graph is a logical execution plan while a DAG is a physical execution plan. DAG consists of various stages, to optimize the execution plan by reducing the data shuffling between the nodes. In Spark, the DAG Scheduler is responsible to implement these stages of DAG. The following steps show how the DAG Scheduler implements these stages:

● Spark creates the Lineage Graph, which is the logical execution plan when the code is submitted.

● The Lineage Graph is submitted to the DAG scheduler by Spark to prepare the physical execution plan.

● The DAG Scheduler splits the graph in various stages, which are grouped together based on the transformations. All the narrow transformations are pipelined together as the first stage and all the wide transformations, which require shuffling, are grouped as the next stage.

● Subsequently, the stages are passed on to the Task Scheduler, which launches the tasks by contacting the cluster manager.

● Finally, the tasks are executed by the executors to produce the final result.

Let us understand this concept with an example. Consider a sample Spark code where we load a file to create an RDD by using textFile API and apply a map function to transform that RDD. Then we create a new RDD by using parallelize. We now have two RDDs. We join them by using the join function and filter the data by using the filter function. Finally, the RDD is stored to the disk by using the saveAsTextFile API.

4(c) Stages in DAG

● The first two operations are narrow transformations, which include reading the file by using textFile API and map function. These operations do not require the data to be shuffled. Therefore, this is considered as Stage 0.

● The next stage is Stage 1, where we use parallelize to create a new RDD.

● After creating the RDD, the next operations are wide transformations, which include a join function to join two RDDs. Actually, we are joining the two RDDs we had created from the textFile and parallelize. After joining, we also perform a filter operation and save the file using saveAsTextFile API. These operations require the data to be shuffled across the cluster from various nodes. This stage will be called Stage 2.

Stage 0 and Stage 1 are executed in parallel, as they are not dependent on each other's outcome. However, Stage 1 can be processed without having to wait for the outcome of Stage 0. The data required to fulfill the operations in Stage 0 and Stage 1 is not shuffled from the other nodes and so the tasks within these stages are executed in parallel. However, Stage 2 is dependent on the outcome of both Stage 0 and Stage 1 and hence it can only be executed after both these stages have provided the result. The data is shuffled from the other nodes across the cluster to fulfill the join operation. Finally, the operations in sequence after the join operation are executed one after the other in the same stage as they are dependent on the outcome from the operations prior to them and provide the final result.

This concludes the theory part of Chapter 4.

AIM

The aim of the following lab exercises is to start writing Spark code and execute it within IntelliJ IDEA, and look at the Spark web interface.

The labs for this chapter include the following exercises:

● Creating a new package in IntelliJ IDEA

● Spark Program – Loading Data

● Spark Program – Performing Operations

● Spark Program – Saving Data

● Spark Program – Lineage Graph

● Spark Web Interface

We need the following packages in order to perform the lab exercise:

● Java Development Kit (JDK)

● Scala

● Spark

LAB EXERCISE 3: SPARK PROGRAM

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

1. Creating a new package in IntelliJ IDEA

2. Spark Program – Loading Data

3. Spark Program – Performing Operations

4. Spark Program – Saving Data

5. Spark Program – Lineage Graph

6. Spark Web Interface

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Task 1: Creating a new package in IntelliJ IDEA

Step 1: Open Intellij IDEA, which we had learned to install 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 repeatedly.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

We can see an interface as depicted in the screenshot below.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

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.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

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

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

Task 1 is complete!

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Task 2: Spark Program – Loading Data

Step 1: Let us now write Spark code in IntelliJ IDEA to count each word in a given file. Right-click on the package, which you had created in the previous step and hover over New and then click on Scala Class.

Step 2: You will be prompted with a pop up manu, which will ask you to enter the name of your Scala class. Please enter wordCount and then click on the dropdown for Kind and select Object as depicted in the screenshot below.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

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

You should see that the name of the package and the object should be pre-populated in the IDE editor.

Step 4: Now that we have an editor for writing the Spark code, the first thing we need to do is to include the import statements from the Spark libraries. This program requires the following imports:

The first two import statements are utilized to import the Spark packages and the last import statement is used to set the logging level for our Spark application.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Please ensure that you enter the import statements above the object line as shown in the screenshot above.

Step 5: Once we have the required imports, we can write the main function similar to that of Java. This is the starting point for the compiler to execute our program.

def main(args: Array[String]): Unit = {

The main function takes an argument args, which is an Array of String type and does not return anything. Moreover, Unit represents no return value similar to void in Java. It is optional to mention the return type.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Step 6: Let us declare the level of logging in order to log the error only and the fatal messages.

Logger.getLogger("Org").setLevel(Level.ERROR)

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

This step is not mandatory and we can skip it if we want all the logs.

Step 7: Let us now create a SparkContext object, so that we can access all the Spark functionality.

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

We are creating an immutable variable called sc, which contains the SparkContext object. Inside the SparkContext object, the first parameter tells Spark if we would want the program to be executed in local or in a distributed mode. In our case, since we are working locally, we will be using local[*]. The [*] tells Spark to utilize all the CPU cores available locally in our machine. The next parameter is just the name of our app, which is the WordCount.

Please note that IntelliJ IDEA automatically prepends the names of the parameters based on what we enter.

Step 8: Now, we have created an object named SparkContext. We can now utilize this object and load data using the textFile API as we had done in the Spark Shell.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Please download a text file from the URL below and save it in the path IdeaProjects/Spark/chapter_4/treasure_island.txt. Please create new directories as required, and the IdeaProjects folder is present in your Home folder.

treasure_island.txt - http://bit.ly/2LBFLtt

Once you have the file downloaded and saved in the desired location, write the following line of code to load the file to create an RDD.

val data = sc.textFile(“chapter_4/treasure_island.txt”)

With this process, we have successfully created an RDD using the text file.

This step completes the process of creating a SparkContext object and creating the first RDD by loading the data using the textFile API.

Task 2 is complete!

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Task 3: Spark Program – Performing Operations

In the previous task, we successfully created an RDD. Now let us use the RDD and apply operations to count the number of words in a file.

Step 1: We have an RDD, which contains text in lines. Let us split the lines to words by using the flatMap function. This function is used to remove a level of nesting. The flatMap function is a combination map and flatten functions, where the map function is applied first followed by the flatten function.

val words = data.flatMap(line => line.split(" "))

The above piece of code splits each line into a separate word. The logic applied to split the line is by a white space character. The flatMap function takes the data RDD and splits each line of word by a space character.

As a lab challenge, apply a map function to see how the output looks like instead of a flatMap function. After you see the output from the map function, apply the flatten function to compare the result of flatMap function and the result of map and flatten function individually.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Step 2: At this point, we have each word in a row. In order to count the occurrences of each word, we have to map it to a key-value pair where the key is the word itself and the value will be number 1.

val wordskv = words.map(word => (word, 1))

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Here we utilize a map function to create a key-value pair for each word where the key is the word and value is the literal number 1. With this operation, we end up having a tuple of word and the literal number 1.

Step 3: Now that we have tuples, all we need to do is to add the values (literal number 1) for the same key. To achieve this, we utilize the reduceByKey function. As the name suggests, the reduceByKey function takes a key and applies the operations to its values.

val count = wordskv.reduceByKey((x,y) => x + y)

The above line of code takes the wordskv RDD and applies the reduceByKey function to perform a sum of all the values for a key. In this manner, we end up with a tuple where the first element is the word and the second element is the number of occurrences for that word.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

The reduceByKey function is similar to the reduce function which we learned in the previous chapter. The difference is that the reduceByKey function performs the reduce operation on values for a given key in a tuple while reduce function is applied for all the elements in the collection.

Step 4: In the previous step, we completed all the transformations. Let us now perform an action to print out the result to the console by using the following line:

count.collect.foreach(println)

We can now simply use collect to collect the final RDD, which is count, and use foreach with println to print out each record in the RDD to the console. This actually triggers the program to evaluate. All the transformations before this action are only logged in the Lineage Graph to achieve the lazy evaluation.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

This completes our first-ever Spark program. All we need to do is to execute it.

Step 5: To run this program within the IDE, simply click on the play button and Run program as depicted in the screenshot. Subsequently, the program is compiled and executed.

Once we click on the play button, the IDE starts executing the program by first compiling and then displaying the result at the bottom. The execution may take some time based on the hardware configuration of your machine. Once the program has been executed successfully, you should see exit code: 0 as shown in the screenshot below. If it shows exit code: 1, there might be a mistake somewhere in the program. Please go back, correct it and then execute the program again.

To view the result, scroll up the console until you see text in white as shown in the screenshot. Following is the output result showing each word in the text file with the number of its occurrences.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Task 3 is complete!

Task 4: Spark Program – Saving Data

The output displayed in the console is useful during the development process, but in the real-time scenarios, we would want the output to be saved.

Step 1: We shall be using the saveAsTextFile API to save the output of our program to the disk. Simply replace the collect statement from the previous task with the following line of code:

count.saveAsTextFile(“chapter_4/word_count/output”)

In this way, we can save our output to the following path IdeaProjects/Spark/chapter_4/word_count/output. We need not create the directories word_count and output, as they are automatically created. In fact, the compiler throws an error if the output directory is already present.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Step 2: Now execute the program as you did in the previous task and check the output directory. You should see two files: part-00000 and a _SUCCESS file. The output is saved in part-00000 file.

Open the part-00000 file in a text editor and you should see the result as depicted in the screenshot below.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Task 4 is complete!

Task 5: Spark Program – Lineage Graph

Let us now check the Lineage Graph for our Word Count program.

Step 1: To check the Lineage Graph for our Word Count program, we should use the toDebugString method. To do so, simply replace the saveAsTextFile line from the previous task with the following code:

count.toDebugString.foreach(print)

Please note that we have utilized print inside foreach and not println.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Step 2: Run the program as you did before and you should see the output as shown below.

As we can observe from the screenshot above, the toDebugString method displays the Lineage Graph. The indentations in the last four lines specify the shuffle boundary. It means that there was no shuffle of data for these operations: map, flatmap, and teftFile. On the other hand, the reduceByKey operation involves shuffling of data.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

The number inside the parenthesis denotes the number of parallel tasks. In our case, it is only 1. The higher number denotes a higher level of parallelism.

Task 5 is complete!

Task 6: Spark Web Interface

We can access the Spark web interface to monitor the execution of Spark applications through a web browser. The web interface can be accessed by navigating to the following URL:

http://[driverHostname]:4040

The driverhostname is usually an IP address in the real-time environment and 4040 is Spark's port by default.

Step 1: Open a terminal and start the spark-shell by entering the following command:

$ spark-shell

The Spark shell shows you that the web interface is available at the following URL as shown below:

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Your driverHostname might be different. If a port is being utilized by another application, Spark increases the port by 1 until an open port is found. For example, if 4040 is already occupied, it will increase the port number to 4041.

Step 2: Now open your web browser and navigate to the web interface URL displayed in your Spark Shell. You can see the Spark web interface as depicted in the screenshot below.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Since there is no job running, you will be unable to see any metrics.

Step 3: Let us run a job. Create a List of few numbers and create an RDD from that list as shown below.

scala> val num = List(1, 2, 3, 4)

scala> val numRDD = sc.parallelize(num)

Now let us write a map function, which takes the numRDD and gives a squaredRDD as shown below.

scala> val squaredRDD = numRDD.map(x => x * x)

scala> squaredRDD.foreach(println)

After you see the output in the console, navigate back to the browser and refresh the Spark web interface. Thereafter, you will see a completed job as depicted in the screenshot below:

Step 4: You can click on the collect link below the Description column and you will be taken to stages. Click on the collect link again to check more information as shown in the screenshot below.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Step 5: Click on the DAG Visualization link in order to view the DAG.

Package in IDE  >> Loading Data  >>  Perform Ops

Saving Data

Lineage Graph

Web Interface

Step 6: Click on the Executors link in the navigation bar to monitor the executors as shown below:

Task 6 is complete!

SUMMARY

In this chapter, we have learned the Characteristics of RDD extensively. Moreover, we have described the types of operations we can apply on RDD. Furthermore, we have comprehensively studied the Spark architecture by discussing the Lineage Graph and DAG.

In the labs, we wrote our first Spark program in Intellij IDEA and learned to execute it. In addition, we have also looked at the Spark web Interface.

REFERENCES

● https://spark.apache.org/

CHAPTER 5:

RDD KEY-VALUE PAIRS

& CACHING

Theory

In the previous chapter, we learned the internal execution of a Spark job and executed our first Spark program through the IntelliJ IDEA. In this chapter, let us learn about RDDs of key-value pairs.

Paired RDD

An RDD containing key-value pairs is known as Paired RDD. They are the most frequently-used data structures in Spark to perform aggregations. Moreover, they are beneficial during the joins where two RDDs are joined based on the same key. In Task 3 of the previous lab exercise, we have already learned to create a Paired RDD. We have used a map function to map each word to a key-value pair where the key is the word itself and the value is number 1. Subsequently, we applied reduceByKey function, which can only be applied on a Paired RDD.

The simplest way to create a Paired RDD is by using the map function as observed in the previous lab exercise. All we need to do is map an RDD as a tuple and Spark will automatically consider the tuple as a key-value RDD or paired RDD. Next, we can apply transformations to the key-value pair RDD as per the requirement. A paired RDD can have a complicated data structure associated with it as well. We can utilize the nesting within the tuple for a key and value. All it needs to have is a key and a value at the top level to be qualified as a Paired RDD.

Paired RDD Transformations

The functions, which are utilized for a standard RDD, can also be employed for Paired RDD. However, since the Paired RDDs are in the form of tuples, we need to apply functions that deal with a key-value pair rather than individual elements. The following are the few transformation functions, which can be applied on the paired RDDs.

reduceByKey

The reduceByKey function is used to combine all the values of a given key using an associative function. The syntax for the reduceByKey function is rdd.reduceByKey((x,y) => x + y). The x and y in the syntax above do not represent a key-value pair. Instead, they represent two values of a given key on which the commutative and associative functions are applied at a time. The result of the first two values of the key is then combined with the third value for the key and so on for all the values of a given key. The reduceByKey function is a wide transformation, as it requires shuffling of data across the nodes of a cluster.

sortByKey

The sortByKey function is utilized to sort all the values of a given key. The syntax for sortByKey is rdd.sortByKey().

groupByKey

The groupByKey function is similar to that of reduceByKey function. It is used to group all the values of a given key in the form of an iterator. The syntax for groupByKey is rdd.groupBykey(). The groupByKey function is also a wide transformation, as it requires shuffling of data across the nodes of a cluster.

mapValues

The mapValues function is used to transform all the values and keep the key as-is without changing. It is only applied on the values. Therefore, the mapValues function is used to transform the values only and the key is not transformed. The syntax for the mapValues function is rdd.mapValues().

flatMapValues

The  flatMapvalues function is similar to that of the mapValues function but it also applies the flatMap function over the mapValues function. It is a combination of the mapValues and flatMap functions. The flatMap function is useful to break down the collections into elements of the collection after the mapValues function is applied.

keys

The keys function is used to return a new RDD of all the keys from the Paired RDD. The syntax for keys is rdd.keys.

values

The values function is used to return a new RDD of all the values from the Paired RDD. The syntax for values is rdd.values.

Two Paired RDD Transformations

Like the single-paired RDD transformations, we can also perform transformations on the two-paired RDDs. An example of such two paired RDD transformation is join. The following are functions that can be performed with two paired RDDs.

join

The join is similar to that of SQL style join, which is used to join two paired RDDs based on a common key. This join action performs an inner join on the two-paired RDDs. The syntax of join is rdd1.join(rdd2).

rightOuterJoin

Given the two-paired RDDs, rdd1 and rdd2, the rightOuterJoin function is used to join these two RDDs where the key is present in rdd1. The syntax of rightOuterJoin function is rdd1.rightOuterjoin(rdd2).

leftOuterJoin

Given the two-paried RDDs, rdd1 and rdd2, the leftOuterJoin function is used to join these two RDDs where the key is present in rdd2. The syntax of the leftOuterJoin function is rdd1.leftOuterjoin(rdd2).

subtractByKey

The subtractByKey function is used to produce a new RDD by removing the elements, which have the same key on both the paired RDDs. The syntax of subtractByKey is rdd1.subtractByKey(rdd2).

cogroup

The cogroup function is used to group data from both the paired RDDs, which have the same key. The cogroup function is similar to that of Full Outer Join in SQL but instead of flattened result of each line per record, we get an iterable interface with cogroup function. Subsequently, we can either tokenize the records or perform other operations as per the requirement. The syntax for cogroup function is rdd1.cogroup(rdd2).

Paired RDD Actions

In addition to transformations, we also have actions that can be applied on paired RDDs. These actions, as observed earlier, trigger the evaluation of a Spark job. All the regular actions, which are applied on RDDs, can be applied on the paired RDDs.  Let us look at the actions for paired RDDs.

countByKey

The countByKey function gives back an RDD with the count of values for each key. This function is very useful when we just need to count the values for each key in the paired RDD. The syntax for the countByKey function is rdd1.countByKey().

collectAsMap

The collectAsMap function is similar to the collect function, but collectAsMap goes a step ahead to return the results for paired RDD as a Map collection. Since this will be a map collection, all the duplicate keys are removed and the result contains the paired RDD with unique keys. Please be informed that when we use the collectAsMap function similar to collect, all the data is shuffled to the driver. We only use this function, when we know that we have enough driver memory to accommodate all the data. The syntax for collectAsMap is rdd.collectAsMap().

lookup(key)

The lookup(key) function is used to look up values corresponding to a key. This provides an efficient way to get values based on the key. The syntax for the lookup(key) function is rdd.lookup(key).

RDD Caching and Persistence

RDD Caching and RDD Persistence play very important role in processing data with Spark. With caching and persistence, we are able to store the RDD in-memory, so that we do not have to recompute or evaluate the same RDD again, if required. This is an optimization technique, which helps to complete jobs more quickly by saving the evaluation of RDD in memory. We learned this characteristic of RDD in the previous chapter.

Let us understand this better with an example. The default behavior is that an RDD is computed every time an action is called on the RDD. Look at the following piece of code below.

scala> val data = sc.textFile(“/some/path/records.txt”)

The above line simply loads a text file using the textFile API and stores it to an RDD called data.

scala> data.take(5)

The above code utilizes the take() function to return first five elements of the RDD. This is an action, which triggers the evaluation. The RDD data is now computed by loading it from the file system and then the action is performed.

Now, let us say that we need to count the elements in our RDD.

scala> data.count()

We are now running a new action, which is causing the RDD to compute again by loading it from the file system and then the action count is performed. As we can observe, we are evaluating the RDD twice. This takes lot of time if the data volume is very big. To overcome this problem, we have the cache() and persist() methods which can be cache or persist the RDD in memory or on the disk.

The difference between cache() and persist() methods is that the cache() uses the default storage level of StorageLevel.MEMORY_ONLY, while the persist() method can have the combination of various storage levels as observed below.

Persistence Storage Levels

MEMORY_ONLY

This is the default storage level. The RDD when cached is stored in memory only. If the RDD does not fit in the memory, few partitions, which do not fit, are computed on the fly when an action is called. The RDDs are stored as the  deserialized Java objects.

MEMORY_AND_DISK

This storage level uses the disk to store few partitions of RDD if they do not fit in the memory. Therefore, instead of recomputing the RDD partitions, which do not fit in memory, they are spilled to disk. The RDDs are stored as deserialized Java objects.

MEMORY_ONLY_SER

This storage level is the same as MEMORY_ONLY but RDDs are stored as serialized Java objects. Serialization is more space efficient when it is compared to the deserialized objects; however, it is CPU intensive operation.

MEMORY_AND_DISK_SER

This storage level is same as MEMORY_AND_DISK but RDDs are stored as serialized Java objects in memory. The partitions, which do not fit in the memory, are spilled to the disk.

DISK_ONLY

In this storage level, the RDDs are stored to the disk only and not in memory. This requires low space when compared to persisting in memory but is CPU intensive.

MEMORY_ONLY_2,

MEMORY_AND_DISK_2,

MEMORY_ONLY_SER_2,

MEMORY_AND_DISK_SER_2,

DISK_ONLY_2

All these levels are same as above but they store the RDD partitions with the replication factor of 2. It means that each partition is stored on two nodes of a cluster with replication.

Let us go back to our example and observe how we can use cache() and persist() methods.

scala> val data = sc.textFile(“/some/path/records.txt”)

Once we load the file using the TextFile API, we can now cache or persist the data RDD. Before we can cache or persist, we have to import the following.

scala> import org.apache.spark.storage.StorageLevel

Subsequently, we use the cache() method, if we need the default implementation of storage only.

scala> data.cache()

However, if we want to use the various storage levels as explained above, we have to use the persist() method and specify the desired storage level. Therefore, the code looks like:

scala> data.persist(StorageLevel.MEMORY_AND_DISK_SER)

At this point of time, we have simply specified out storage level for persistence. The actual persistence happens when the action on the RDD is called.

scala> data.take(5)

After this action is completed, the RDD is stored in the memory and any partitions that do not fit in memory are spilled to disk. When we trigger another action as below, the RDD will not be computed again as it is already computed and persisted. This process reduces the total time taken to complete the job without having to compute the same RDD repeatedly.

scala> data.count()

It is also possible to remove the persisted RDDs manually. We simply have to use the unpersist() function to the RDD we want to unpersist.

scala> data.unpersist()

However, if we choose not to remove the persisted RDDs manually, Spark automatically removes the partitions based on the Least Recently Used (LRU) cache policy, when there is too much data cached in memory. Therefore, we need not worry about breaking a job when memory is full.

At this stage, we conclude the theory for this chapter.

AIM

The aim of the following lab exercises is to start writing Spark code in Intellij IDEA to learn about the Paired RDDs.

The labs for this chapter include the following exercises:

● Creating a Tuple

● Creating a Paired RDD

● Performing operations on Paired RDD

● Performing more operations on Paired RDD

● Performing Joins on Paired RDD

● Performing Actions on Paired RDD

We need the following packages to perform the lab exercise:

● Java Development Kit (JDK)

● Scala

● Spark

LAB EXERCISE 4: PAIRED RDD – HANDS ON

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

1. Creating a Tuple

2. Creating a Paired RDD

3. Performing operations on Paired RDD

4. Performing more operations on Paired RDD

5. Performing Joins on Paired RDD

6. Performing Actions on Paired RDD

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Let us start this exercise by executing a program, which computes the average rating provided by a user for a list of movies.

Task 1: Creating a Tuple

Step 1: Download the ratings.csv file from the URL below. This file contains four columns: userId, movieID, rating and timestamp.

ratings.csv - http://bit.ly/2QmnAH9

Please save this file in the IdeaProjects/Spark/chapter_5 folder. (Please create a folder named chapter_5 in the home folder.)

Step 2: Open IDE, right-click the training package that you had created in previous exercise and hover over New and then click on Scala Class. When prompted, enter avgRatings as the name, and click on the dropdown for Kind and select Object. Enter the imports as depicted below.

import org.apache.spark._

import org.apache.spark.SparkContext._

import org.apache.log4j._

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 3: Let us now write a function to parse the records and extract the fields of our interest. For our program, we are only interested in the userId and ratings fields. The function will split a line of input into a tuple of (userId, ratings).

Let us name the function parseRecords.

def parseRecords (rows: String): (Int, Float)={

This function takes each line of input as an argument and returns a tuple of an integer and a float.

Step 4: Now, using the split function, we split each field in the record by a comma, as we know that each field in our record is separated by a comma.

val records = rows.split(",")

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 5: Now that we have split the records, all we have to do is to extract the required fields and convert the userId to integer and ratings to float types.

val userId = records(0).toInt

val ratings = records(2).toFloat

The records variable contains four fields. We can simply access them based on the index starting from 0. Therefore, we simply extract the userId which is the first field to the variable userId and ratings which is the third field to the variable ratings.

Step 6: Finally, we simply have to return the userId and ratings to complete our function.

(userId, ratings)

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

This creates a tuple of userId and ratings. We can use this tuple to create our Paired RDD in the next task.

Task 1 is complete!

Task 2: Creating a Paired RDD

Let us now write the main function for our program where we create our Paired RDD and perform operations over the Paired RDD.

Step 1: Write the following main function and error log level as shown below.

def main(args: Array[String]): Unit = {

Logger.getLogger("Org").setLevel(Level.ERROR)

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 2: Create a SparkContext object as we had learned in the previous exercise. Enter the master as local to use all the cores and the name of the app as Average ratings by users.

val sc = new SparkContext("local[*]", "Friends By First Name")

Now that we have the SparkContext object created, let us load our file using the textFile API.

val data = sc.textFile(“chapter_5/ratings.csv”)

We now have an RDD loaded.

Step 3: Finally let us create our pairedRDD. To do this, we have to pass the parseRecords function as an argument to the map the higher order function, so that all the records in the data RDD are parsed as per the logic in our parseRecords function. The following line of code performs this task.

val RDDPair = data.map(parseRecords)

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

That is it. We now have our Paired RDD. We can optionally look at the RDDPair by simply printing it out to the console by using the code below.

RDDPair.collect.foreach(println)

PS: Using collect is not recommended if your data volume is very big. When collect is used, all the data is shuffled to the driver node and if there is not enough memory available in the driver node, the job will throw a memory exception error.

Once you write this line of code, run the program by clicking the play icon and then clicking on the Run option as shown below. Please ensure that you have the appropriate closing flower braces at the end of the code.

You should have the output in the console with the key-value pairs as shown in the screenshot below.

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

The first element in each tuple is the key (userId) and the second element in the tuple is a value (ratings).

Task 2 is complete!

Task 3: Performing Operations on Paired RDD

Before proceeding with this task, please comment out or remove the following statement from the previous task.

RDDPair.collect.foreach(println)

We can put comments in Scala using the characters ‘//’ preceding the comment as shown below. These kinds of comments are called single line comments.

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

We can also use the multi-line comments by using characters ‘/*’ and ‘*/’ around the comment.

/*

This is a multi-line comment.

Which can span multiple lines.

Like this.

*/

Step 1: At this point, we have a Paired RDD with userId as key and ratings as value. We now have to compute the sum of user ratings and divide them by the number of ratings, so that we can get the average rating by a user.

To achieve this, we must first compute the number of ratings by a user using the mapValues function.

val mappedRatings = RDDPair.mapValues(x => (x,1))

This transformation converts each rating value to a tuple of (ratings, 1). Therefore, we will be having our Paired RDD as (userId, (ratings, 1)).

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

You may optionally print out the mappedRatings to the console as in the previous task, to check how the result is displayed.

Step 2: Next, we shall be using the reduceByKey function to sum up all the ratings and all the instances for each userID, by adding all the rating values and 1's, respectively.

val totalRatings = mappedRatings.reduceByKey( (x,y) => (x._1 + y._1, x._2 + y._2))

The result is in the form of (userId, (totalRatings, totalInstances))

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 3: Finally, we can now compute the average of the ratings by userId again by using the mapValues function. The average is calculated by dividing totalRatings by totalInstances.

val avgRatings = totalRatings.mapValues(x => x._1/x._2)

We now have the average rating of each user.

Step 4: We can now use the collect function to compute the final result from the RDD and display it on the console using the following line of code. This action triggers DAG and the job is executed.

avgRatings.collect.foreach(println)

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 5: Run the program by clicking on the play icon and Run option as depicted in the screenshot below.

Once the job is finished, check the output in the console as shown in the screenshot below.

Step 6: We can also sort the result on either column by using the sortBy function as shown below.

val sorted = avgRatings.sortBy(x => x._2)

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

The above line is utilized to sort the second field, which is the value (Average rating) in the ascending order by default. The sorted result is as shown below.

However, if we want to sort it in descending order, we can simply use the dash (-) symbol as shown below.

val sorted = avgRatings.sortBy(x => -x._2)

We will have the results sorted in descending order when we run the program as shown below.

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

We can also sort the result by the key by referring to the first element in the sortBy function or by simply using the sortByKey function key as shown below.

val sorted = avgRatings.sortByKey()

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Finally, we can sort the results in the descending order when the sortByKey function is used by passing the argument as false for the function.

val sorted = avgRatings.sortByKey(false)

Task 3 is complete!

Task 4: Performing more Operations on Paired RDD

In the previous tasks, we have only performed a couple of operations on the Paired RDDs. Let us now look at some more operations that we can perform on Paired RDDs.

Step 1: Download the tags.csv file from the URL below. This file contains four columns: userId, movieID, tag and timestamp.

tags.csv - http://bit.ly/2YTVGFk

Please save this file in the IdeaProjects/Spark/chapter_5 folder. (Please create a folder named chapter_5 in the home folder.)

Step 2: Open IDE, right-click the training package that you have created in the previous exercise and hover over New and then click on Scala Class. When prompted, enter tags as the name, and click on the dropdown for Kind and select Object. Enter the required imports as depicted in the screenshot below.

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 3: Write the recordsParser function as in the previous task. For this task, let us extract the movieID and tag fields. The recordsParser function is as shown below.

def parseRecords (rows: String): (Int, String)={

val records = rows.split(",")

val movieID = records(1).toInt

val tags = records(2).toString

(movieID, tags)

}

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 4: Create a Paired RDD as in Task 2 by writing the main function, and setting the error log level (optional), creating a SparkContext object and loading the file using the textFile API.

def main(args: Array[String]): Unit = {

Logger.getLogger("Org").setLevel(Level.ERROR)

val sc = new SparkContext("local[*]", "Paired RDD Operations")

val data = sc.textFile(“chapter_5/tags.csv”)

Now create an RDD pair by parsing the data RDD using the recordsParser function.

val RDDPair = data.map(parseRecords)

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

We now have our paired RDD. Let us use some operations in the next step on our Paired RDD.

Step 5: Now that we have our Paired RDD, let us group all the tags by movieID using the groupByKey function.

val grouped = RDDPair.groupByKey()

Let us now print out the result of the grouped RDD to the console.

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

grouped.collect.foreach(println)

The output is as shown in the screenshot below with all the tags for a movie grouped together.

We may optionally convert the values from compactBuffer to a list by simply mapping the output and converting them to a List as shown below.

val flattened = grouped.map(x => (x._1, x._2.toList))

flattened.collect.foreach(println)

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 6: We can also extract the keys and values to separate RDDs as shown below.

val RDDKeys = flattened.keys

RDDKeys.collect.foreach(println)

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Similarly, we can extract the values by using the code below.

val RDDValues = flattened.values

RDDValues.collect.foreach(println)

Task 4 is complete!

Task 5: Performing Joins on Paired RDDs

So far, we have worked on single paired RDDs. In this task, let us look at two paired RDDs by performing joins.

Step 1: Download the ratings.csv file from the URL below. This file contains four columns: userId, movieID, rating and timestamp.

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

ratings.csv - http://bit.ly/2QmnAH9

Please save this file in the IdeaProjects/Spark/chapter_5 folder if not saved already in the previous tasks.

Step 2: Download the movies.csv file from the URL below. This file contains three columns: movieID, movieName and genre.

movies.csv - http://bit.ly/2EJj0Os

Please save this file in the IdeaProjects/Spark/chapter_5 folder.

We shall join these datasets based on the movieID.

Step 3: Create a new object in the IDE and name it joins. Import all the required import statements as shown below. Next, let us declare a case class with fields according to the columns in both the files, along with their data types.

import org.apache.spark._

import org.apache.log4j._

object joins {

case class ratings(userId: Int, movieID: Int, rating: Float, timestamp: String)

case class movies(movieID: Int, movieName: String, genre:String)

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 4: Now, let us load both the files using the textFile API and split the fields by a comma delimiter. We shall be using the map function to split the fields for every record in the RDDs.

def main(args: Array[String]): Unit = {

Logger.getLogger("Org").setLevel(Level.ERROR)

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

val rating = sc.textFile("chapter_5/ratings.csv").map(x => x.split(','))

val movie = sc.textFile("chapter_5/movies.csv").map(x => x.split(','))

By splitting the fields, we can refer them individually while creating a Paired RDD.

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 5: Now that we have both the datasets loaded and separated by comma delimiter, let us create a tuple of two elements (Paired RDD) so that we can perform a join on both the RDDs based on key. The key here is the movieID.

To create a tuple, we utilize the map function with the first element (key) of the tuple as movieID, which is second field in the rating RDD and first field in movie RDD. The second element (value) of the tuple will be the entire records from both the RDDs.

val rating_record = rating.map(x => (x(1).toInt, ratings(x(0).toInt, x(1).toInt, x(2).toFloat, x(3).toString)))

val movie_record = movie.map(x => (x(0).toInt, movies(x(0).toInt, x(1).toString, x(2).toString)))

Step 6: The next step is to perform the join as shown below.

val joined = rating_record.join(movie_record)

Now we can simply collect the results and print them to the console.

joined.collect.foreach(println)

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

The output with the joined results can be observed in the screenshot below.

Similarly, we can perform the rest of the joins, such as left outer join and right outer join.

It is recommended to perform joins using the dataframes rather than RDDs as it can have the benefit of catalyst optimizer when performed using dataframes. We shall look at dataframes in our upcoming chapters.

Task 5 is complete!

Task 6: Performing Actions on Paired RDDs

Let us now perform actions on the Paired RDDs and look at caching and persisting RDDs. In addition, let us continue the program from the previous task and apply actions over them.

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 1: We shall be using the joined RDD for all the actions we perform in this task. Since we shall be using this RDD more than once, let us persist with it before performing these actions, so that the application does not process all the transformations repeatedly whenever an action is called.

joined.persist(StorageLevel.MEMORY_AND_DISK_SER)

We might see an error about the missing import. If so, please add the following import to the list of imports.

import org.apache.spark.storage.StorageLevel

We may use our desired level of persist storage level above.

Step 2: Let us first use the countByKey function to check the number of ratings per movie on our joined RDD.

val count = joined.countByKey()

Let us print the count to the console using the println function and run the program. We should see the result as shown below with count of reviews for each movie shown as a Map collection.

println(count)

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

Step 3: Let us now use the collectAsMap function on the joined RDD.

val mappedCol = joined.collectAsMap()

println(mappedCol)

The result is returned as a Map collection with all the duplicate keys removed.

Step 4: Finally, let us utilize the lookup(key) function to lookup value for a key in our joined RDD.

Creating Table  >> Creating RDD Pair  >>  Pair RDD Op

More Op.

Pair RDD Joins

Pair RDD Actions

val look = joined.lookup(25)

println(mappedCol)

The result is shown as an ArrayBuffer for all the values of the key.

Task 6 is complete!

LAB CHALLENGE

● As a continuation to Task 5, please perform the other joins (Right Outer and Left Outer joins), which we have learned in theory.

SUMMARY

In this chapter, we have comprehensively studied the RDD key value pairs. We have learned the types of operations we can apply on Paired RDDs; moreover, we discovered the caching and persistence.

In the labs, we created a Paired RDD and applied transformations and operations on the Paired RDDs.

REFERENCES

● https://spark.apache.org/

CHAPTER 6:

SHARED VARIABLES

Theory

Spark provides two different abstractions as low-level APIs. We have been looking at RDDs, which are the low-level Application Programming Interfaces (APIs) for Spark so far. In this chapter, we shall be looking at the Distributed Shared Variables, which are also other types of low-level abstractions for Spark. Let us learn what are the Distributed Shared Variables and why we should use them.

What are Shared Variables?

The Shared Variables in Spark are the variables, which are used to perform operations in parallel. When a job is run in Spark, the functions of that job are run in parallel as tasks on various executors of the cluster. The variables used in the functions are sent to each task each time the variable is required. Each task gets a copy of the variable in serialized form and changes made to these variables are not propagated back to the driver or to other tasks in executors processing them. This default behavior works well when the requirement does not have to manipulate the variables or if the variables are shipped only once. However, this is not the case all the time and is inefficient to ship a copy of variable and deserialize it every time it is required during the job, especially if the value is considerably large.

Why Shared Variables?

As explained in the previous section, in order to overcome the inefficiency of shipping the variables every time to executors, and to overcome the inability of propagating the updates of variables back to the driver or executors, we use the Distributed Shared Variables.

There are two types of Distributed Shared Variables supported by Spark.

● Broadcast Variables

● Accumulators

Let us look at each of them in detail and try to understand them better with an example in the lab exercises.

Broadcast Variables

Broadcast variables are the shared variables, which allow Spark to send large values efficiently in an immutable (read-only) state to all the worker nodes. These variables can be utilized one or more times during Spark operations. The broadcast variables are sent to the worker nodes only once and are then cached to the worker nodes’ memory in the deserialized form. These variables are very useful when the Spark job consists of multiple stages and multiple tasks in those stages require the same variable. Broadcast Variables overcome the inefficiency of shipping the variables every time to executors.

If not for broadcast variables, the required variable should be shipped to executors running the tasks every time it is referenced, cache it in the executor’s memory and then deserialize it. All this accounts to an overhead and hampers the job performance. Instead of shipping the variable every time it is required during the multiple stages, it can be simply accessed any number of times without any overhead as it is available locally when the Broadcast Variables are used. An example for the Use Case of broadcast is a look-up table or a feature vector in machine learning algorithm.

Using Broadcast Variables does not mean that the data is not transmitted across the network. However, unlike transmitting the data every time the variable is referenced in the “regular” variables, the data is only transmitted once, saving network bandwidth and executor resources.

Optimizing Broadcast Variables

It is imperative to optimize Broadcast Variables with compact and fast data serialization when large values are used to broadcast. Not doing so will result in network bottlenecks, if the value takes too much time to serialize and transmits over the network. Using the correct serialization library helps to improve the job performance in an efficient manner.

The default serialization library, Java Serialization, provides fast and compact serialization for almost everything except arrays of primitive types. We can use a different serialization library to optimize our broadcast using the spark.serializer property, or we can implement our own user-defined serialization as per the requirement.

We shall be looking at an example program using Broadcast variables in the lab exercise for this chapter.

Accumulators

Another kind of the Distributed Shared Variables are Accumulators. They are most commonly used in error handling, i.e., to implement custom counters to count the events that occur during the job execution. Let us learn about Accumulators in detail.

As the name suggests, Accumulators are used to accumulate values to implement counters and aggregations. Since accumulation involves updating the values, they are mutable (writable), unlike the Broadcast Variables, which can be updated by tasks running on various nodes and can be read by the driver program. Accumulators overcome the inability of propagating the updates of variables back to the driver. Spark has the support for numeric type Accumulators by default, but we can also develop Accumulators for other types as well.

We can create an accumulator variable by using the accumulator method provided by the SparkContext object. The accumulator method takes two arguments. The first argument is the initial value for the accumulator, and the second argument is the name we can specify for the accumulator, which is displayed in Spark web UI. The second argument is optional and the method is valid with only one argument, i.e., the initial value for your accumulator. The initial value can be incremented by tasks locally by using the add method or += operator. The driver then globally increments all the accumulated values from tasks. The driver uses the value method to read the accumulated value.

Accumulators are efficient because no data is shuffled across the network. All the data is processed locally in the executor nodes by tasks. However, it is not recommended to utilize the Accumulators when the locally accumulated data is large and does not fit the memory of the driver. The regular RDD transformations should be used in such cases.

Points to remember when Accumulators are used

● Accumulators should be used within the action operations only. If they are used within transformations, there is no guarantee that the final results are accurate. Since transformations are executed lazily whenever an action is called, the Accumulators are processed more than once and so the result would be incorrect.

● Accumulators are evaluated in the driver node since it is not associated with an RDD. The rule of thumb is that any code associated with RDD is evaluated in executors and the code, which is not associated with an RDD, is evaluated in the driver node.

We shall be looking at Accumulators with respect to Spark version 1.x and Spark version 2.x in the lab exercises to understand the concept better. We shall also look at creating a custom Accumulator.

Scala Monadic Collections

Now that we are looking at error handling in Spark, let us discuss the Monadic Collections in Scala, which are used for error handling.

There are three Monadic collections in Scala, which are used in error handling in functional programming, similar to try catch blocks in Java exception handling. However, in functional programming, we have to return something while try catch blocks in Java do not return anything. Therefore, Monadic collections help us to follow the rules of functional programming while implementing error handling.

Unlike the traditional collections such as List, Map, Set etc., which can contain multiple objects, Monadic collections only contain two objects and can only return one object at a time. Let us look at these collections in detail.

Either Monadic Collection

The Either collection contains two objects called Left and Right object. They work in the following manner: we can implement a condition in Either and return the Left object in case of errors if the condition in Either is not satisfied and return the Right object in case of the condition in Either is satisfied. Please remember that Left for errors and Right for no errors is just a convention. We can use vice-versa as well. We shall look at this collection in our lab exercise.

Option Monadic Collection

The Option Monadic Collection helps us deal with the NullPointerException. As explained earlier, Option Monadic Collection also contains two objects called Some and None. Let us create a case class, which stores the employee information.

scala> val welcome: Option[String] = Some(“Welcome to Learning Voyage”)

We can now retrieve the value using the get method as String is wrapped into an option.

scala> welcome.get

Similarly, we can also set the None object to the variable as shown below.

scala> val welcome: Option[String] = None

Since there is no value set for our variable, we can use the getOrElse method to set a value on the fly as shown below.

scala> welcome.getOrElse(“Ernesto Lee Website”)

This is a very basic example of using the Option collection. We can use it in a variety of cases like pattern matching and case class etc.

Try Monadic Collection

The Try Monadic Collection is similar to that of Either which contains two objects, namely Success and Failure objects. As the name suggests, the Success object returns the value if the condition in Try succeeds, else the Failure object is returned with the exception.

Please check the References section to learn more about Monadic Collections in Scala.

This concludes the theory for this chapter.

AIM

The aim of the following lab exercises is to start writing Spark code in Intellij IDEA to learn about Broadcast Variables and Accumulators in 1.x and 2.x as well as custom Accumulators.

The labs for this chapter include the following exercises:

● Using Accumulator Method

● Implementing Record Parser

● Implementing Counters

● Implementing Accumulators V2

● Implementing Custom Accumulators

● Using Broadcast variables

We need the following packages to perform the lab exercise:

● Java Development Kit (JDK)

● Scala

● Spark

LAB EXERCISE 5: SHARED VARIABLES – HANDS ON

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

1. Using Accumulator Method

2. Implementing Record parser

3. Implementing Counters

4. Implementing Accumulators V2

5. Implementing Custom Accumulators

6. Using Broadcast Variables

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Let us start this exercise by looking at Accumulators.

Task 1: Using Accumulator method

In this task and the next two, we shall be looking at Accumulators API in Spark 1.x to count the number of malformed records and separate the malformed records from the good records.

Step 1: Download the ratings-malformed.csv file from the URL below. This file contains four columns: userId, movieID, rating, and timestamp.

Ratings-malformed.csv - http://bit.ly/2WuTese

Please save this file in the IdeaProjects/Spark/chapter_6 folder. (Please create a folder named chapter_6 in the home folder.)

Please note that this file is different from the file we used in the previous lab exercises. This file has malformed records.

Step 2: Open IDE, right-click the training package that we had created in the previous exercise and hover over New and then click on Scala Class. When prompted, enter counters as the name, click on the dropdown menu for Kind, and select Object. Enter the imports as shown below.

import org.apache.spark._

import org.apache.spark.SparkContext._

import org.apache.log4j._

Step 3: Now write the main function along with the error log level setting as always.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

def main(args: Array[String]): Unit = {

Logger.getLogger("Org").setLevel(Level.ERROR)

Moreover, create a SparkContext object and enter the master as local to use all the cores and the name of the app as Counters.

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

The code so far should look like the one in the screenshot below.

Step 4: Now that we have the SparkContext object created, let us load our file by using the textFile API.

val data = sc.textFile(“chapter_6/ratings-malformed.csv”)

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

The aim of this task is to count the number of malformed records. However, what is the process of deciding which records are good and which are malformed? To answer this question, we need to have a look at our input file. Open the file, which you have downloaded in step 1 for this task.

You will be able to see that the file contains four fields as explained in Step 1. However, there are many records, which are missing some fields as depicted in the screenshot below.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

In order to count and separate the good records from bad records, we make use of Accumulators.

Step 5: The next step is to use the Accumulator method in the SparkContext object and pass its arguments. The arguments are the initial value of zero (0) and the name of our accumulator as bad records.

val badRecords = sc.accumulator(0, "bad records")

You will see a warning “Symbol Accumulator is deprecated.” You may ignore this warning, as this is an older Accumulator API for Spark 1.x.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

With this process, we have successfully set the initial value for the accumulator and named the counter as bad records. We can now write a logic on how to determine good and bad records. We can achieve this by writing a recordParser object and defining a parse method within the object. Let us see that in the next task.

We shall come back to this program in Task 3, once we figure out how to separate the bad records from the good records, so that we can count the number of bad records.

Task 1 is complete!

Task 2: Implementing Record Parser

Step 1: To implement a record parser, we shall create yet another Scala object and name it as recordParser. Please follow the steps associated with creating a new Scala object. No imports are required for this object.

We should end up with the object as shown below.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 2: Let us first create a case class to store all our good records as shown below.

case class records(userId: Int, movieId: Int, rating: Double, timeStamp: String)

Step 3: Let us now define a parse function, which takes input record of type String as argument. The return type is an Either Monadic collection, which either returns String as Left object or a records case class as Right object.

Do not worry if you do not understand this as of now. All this makes sense when you look at the rest of the code.

def parse(record:String):Either[String, records]= {

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Next, let us declare an array variable of type String and name it fields, so that we can split the incoming records based on a comma. This way, we can access each field individually, and we can know the number of fields in each record.

val fields: Array[String] = record.split(",")

Step 4: This is the step where we perform the error handling by using an ‘if’ loop. The condition for ‘if’ loop will check if a record has four fields by using the length method.

If there are four fields in a record, we simply access each field based on its index and store it in the case class records by using the Right object. If there are less than four fields, we pass the record as it is by using the Left object.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

if (fields.length == 4)

{

val UserId: Int = fields(0).toInt

val movieId: Int = fields(1).toInt

val rating: Double = fields(2).toDouble

val timeStamp: String = fields(3)

Right (records(userId: Int, movieId: Int, rating: Double, timeStamp: String))

}

else{

Left(record)

}

}

}

Please ensure that you correctly enter all the opening and closing flower brackets if you encounter an error.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Let us now go back to the previous program counters and refer to this object there.

Task 2 is complete!

Task 3: Implementing Counters

Step 1: Navigate back to the counters object in IDE and continue from where we left in Task 1.

Step 2: Let us now write a foreach action, which parses each row through our recordParser object, which we had implemented in the previous task.

data.foreach(row => {

val parsedRecords = recordParser.parse(row)

if(parsedRecords.isLeft){

badRecords += 1

}

})

We then write an ‘if’ condition to check if the record is a left object, i.e. a bad record. If it is, we simply increment the value of badRecords variable, which we had declared in task 1 by 1.

We can use the ‘isLeft’ method to check if it is a Left object and ‘isRight’ method for Right object.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Please note that we are using the foreach function, which is an action and not a transformation. As learned in the theory section, Accumulators should always be specified in the action part and not in the transformations. This way, we can ensure that our accumulator is only processed once and the value is accurate.

At this point, we have successfully implemented the counters based on our requirements. All we need to do now is to get the final accumulated value.

Step 3: Let us use the println function to print the number of bad records in our input dataset.

println("The number of bad records in the input are  " + badRecords.value)

}

}

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

To retrieve the value from our accumulator, which is badRecords, we use the value method. We cannot directly retrieve the value just by using the badRecords variable when it comes to the Accumulators.

Step 4: Finally, let us run our code and check the output. We should see the number of bad records as shown in the screenshot below.

With this process, we have successfully implemented the Accumulators using Spark 1.x API.

Step 5: Remember that we have also wrapped all the good records in the Right object. We can use it in our ‘if’ condition with else object here as well.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

else {

val goodRecords = parsedRecords.right.map(x => (x.userId, x.movieId, x.rating, x.timeStamp))

goodRecords.foreach(println)

}

We are declaring a new variable called goodRecords and simply extracting (map) the fields from the Right object by using the right method. Finally, we can print them out to the console in the next line.

Step 6: Let us now execute the code, and observe the good records as shown in the screenshot below.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

We have successfully implemented Accumulators and separated the good records from the bad records.

Task 3 is complete!

Task 4: Implementing Accumulators V2

Please can observe that this is an implementation of Accumulators in Spark 2.x. There will be some code related to the dataframes, which we have not yet covered. However, we should not be worried. Just look at the implementation of Accumulators V2 and after we cover the dataframes, it all makes sense.

Step 1: We shall be using the same file we used in Task 1 for this task as well, since we are accumulating the bad records in the input data.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 2: Open IDE, right-click the training package that you have created in the previous exercise and hover over New and then click on Scala Class. When prompted, enter countersV2 as the name, click on the dropdown menu for Kind, and select Object. Enter the import as shown below.

import org.apache.spark.sql.SparkSession

Since we are working on Spark 2.x, we have to import a SparkSession object instead of a SparkContext object. The SparkContext object is wrapped within the SparkSession object. The SparkSession object is used to read the data.

Step 3: Let us now write our main method and create a SparkSession object so that we can access Spark functionality. Please note that we have not covered this topic but just think of this as if we are creating a SparkContext object.

def main(args: Array[String]) {

val sparkSession = SparkSession.builder

.master("local[*]")

.appName("Bad record counter V2")

.getOrCreate()

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

We have created a SparkSession object by using the SparkSession.builder method, specified the execution environment as local by using all the cores in our CPU and finally named our application as Bad record counter V2. The getOrCreate method gets an instance of the SparkSession object if it is already available, or otherwise, it creates one.

Step 4: Let us now declare our Accumulator object. In Spark 2.x, it is a bit different from what we had seen in Spark 1.x. There are two types of Accumulators, which we can utilize in Spark 2.x: They are called the longAccumulator and doubleAccumulator. As their names suggest, a longAccumulator is used for Long data type and doubleAccumulator for Double data type.

We shall be using the longAccumulator in our code, as we only need the count of type Long.

val badRecords = sparkSession.sparkContext.longAccumulator("Bad Records")

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

The longAccumulator object is wrapped in the sparkContext object, which in turn is wrapped with the SparkSession object. The initial value for longAccumulator is set to zero (0) by default. We need not initiate it as we did in Spark 1.x API for the Accumulator. All we need to do is to set a name for our Accumulator. We have named it Bad Records. Nonetheless, we can use any name as we like.

Step 5: The next step is to read the input data. Please note that this is a different API in Spark 2.x to read the data, and we have not covered it yet. For now, just think of this as a way to read the input data. Moreover, we also need an import for implicits.

import sparkSession.implicits._

val options = Map("header" -> "false", "InferSchema" -> "true")

val data = sparkSession.read.text(“chapter_6/ratings-malformed.csv”).as[String]

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 5: Now, let us write a foreach statement to take each record from the input data, split it by a comma and test it against a condition. If the records do not contain four fields, we increment the badRecords variable by one using the add method as shown below.

Since we know that our records contain four fields and if there are less than four fields in a record, we treat it as a bad record.

data.foreach(record => {

val fields = record.split(",")

if (fields.size != 4)

badRecords.add(1)

})

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 6: Let us now write a print statement to print out the number of bad records in our input data.

println("The number of bad records in the input are  " + badRecords.value)

}

}

With this, we have successfully implemented Accumulators for Spark 2.x.

Step 7: Finally, let us execute our code. We should see the results as shown in the screenshot below.

Task 4 is complete!

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Task 5: Implementing Custom Accumulators V2

In the previous tasks, we have witnessed how to use Spark’s built-in Accumulators. Let us now look at the steps to implement customized Accumulators in Spark 2.x.

Let us implement a Custom Accumulator, which counts the number of times each movie has been rated. We can achieve the same by using the map transformation and then applying the reduceByKey action. Then this will have the data shuffled across the nodes of the cluster. However, when we use Accumulators, the data is not shuffled across the clusters as each executor processes it locally and has its own local accumulator. The only data shuffled across the cluster is the count from each local accumulator, which is only a few bytes. The local count from all the executors is aggregated by the global accumulator in the driver, thus providing the final result.

Step 1: Download the ratings_head.csv file from the URL below. This file contains four columns: userId, movieID, rating, and timestamp.

Ratings_head.csv - http://bit.ly/2X3r2wb

Please save this file in the IdeaProjects/Spark/chapter_6 folder.

Step 2: Open IDE if not already, right-click the training package that you had created in the previous exercise and hover over New and then click on Scala Class. When prompted, enter CountByMovie as the name of the Class. Please note that we are creating this class for this particular task only, and not an object. Enter the imports as shown below.

import org.apache.spark.util.AccumulatorV2

import scala.collection.mutable.HashMap

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

The first import is version two of the Accumulator. The second import is a mutable HashMap, as we are storing our movies and a total number of ratings as key and value, respectively. In this case, we explicitly import the HashMap collection, or else we would end up having an immutable HashMap when we declare one.

Step 3: We now have to extend our class to inherit AccumulatorV2 and then specify the input and output. The input to our Accumulator would be a tuple (movieId and count), processed by each task (local accumulator) on executors, and the output is a HashMap, which is aggregated by the global accumulator on driver.

class CountByMovie extends AccumulatorV2[(Int, Int), HashMap[Int, Int]]{

You may ignore the red error asking to implement the merge method under the class name for now. This error will be gone once we implement all the methods in the code.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 4: Let us now declare a private HashMap variable called the movieCount, which holds the final count of our CountByMovie Accumulator.

private val movieCount = new HashMap[Int, Int]()

Next, we have to implement a reset method available in the AccumulatorV2 class to reset the accumulator value to zero.

def reset() = {

movieCount.clear()

}

Step 5: Subsequently, we implement the add method to specify the aggregation logic for local accumulators, i.e., the tasks that run on executors. All the tasks running on the executors will run the method to aggregate data locally.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

def add(tuple: (Int, Int)): Unit = {

val movieId = tuple._1

val updatedCount = tuple._2 + movieCount.getOrElse (movieId, 0)

movieCount += ((movieId, updatedCount))

}

The add method takes two arguments as key and value. The key, which is the first argument, is the movieId, and the second argument, the count of the movieId, is value. We simply extract them into their respective variables and add them to the movieCount Hashmap. Moreover, the getOrElse method is used to get the value of count, if it exists, or set a value for that movie as zero, and add them with the current count and previous count to get the updated count.

Step 6: The next step is to implement the merge method that actually aggregates all the values from executors and provides the final count.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

def merge(tuples: AccumulatorV2[(Int, Int), HashMap[Int, Int]]): Unit = {

tuples.value.foreach(add)

}

def value() =  movieCount

When all the tasks complete the execution, the final results from all the executors are then sent to the driver where the merging happens. The merge method takes an AccumulatorV2 as an argument, which takes a tuple and returns a HashMap as output. The merge method is called on all the local accumulators from the tasks processed in the executors. Therefore, we use the add method inside the foreach function.

Since we had declared the HashMap as private, we can only access it through the value method. The value method is used to get the current value in our accumulator.

To summarize, the merge method takes an accumulator as an argument and merges all the local accumulators, which were processed in the executors by tasks, based on the logic in add method, into the global accumulator. Moreover, the value method is utilized to get the current value of the HashMap variable movieCount.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 7: Next, there are a couple of methods required to complete our implementation of the custom accumulator. They are the isZero and the copy methods.

def isZero(): Boolean = {

movieCount.isEmpty

}

def copy() = new CountByMovie

The isZero method returns a Boolean by checking if the accumulator value is zero or not, whereas the copy method is used to create a new copy of our accumulator object.

These are the abstract methods, which must be implemented in our code as they are defined in the base class. They are utilized while aggregating the value in the accumulator.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

The error for the class name should be gone now. With this process, we have successfully implemented our Accumulator V2. We can now use this custom accumulator in our main program.

Step 8: Create a new object by right-clicking the training package, which you had created in previous exercises and hover over New and then click on the Scala Class. When prompted, enter countByMovieMain as the name, click on the dropdown menu for Kind, and select Object. Enter the import as shown below.

import org.apache.spark.sql.SparkSession

Let us first create a case class with all our fields for the input data outside the object as shown in the screenshot below.

case class Movies(userId: Int, movieId: Int, rating : Double, timeStamp: String)

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 9: Let us now write our main function and create a SparkSession object.

def main(args: Array[String]) {

val sparkSession = SparkSession.builder.

master("local[*]")

.appName("Count By movieId")

.getOrCreate()

Next, let us create the Accumulator object and register it using the register method as shown below. We have to register our Accumulator since it is a custom accumulator. However, in case of built-in accumulators, we do not require registration.

val countByMovie = new CountByMovie()

sparkSession.sparkContext.register(countByMovie)

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 10: Let us now write some code to read the input data. We also need to import the implicits.

import sparkSession.implicits._

val options = Map("header" -> "true", "inferSchema" -> "true")

val data = sparkSession.read.format("com.databricks.spark.csv")

.options(cvsOptions)

.load(input)

.as[User]

Do not worry if this code does not any make sense. Just think of this as a way to read the input data, as we used to do with the SparkContext object in the previous exercises. Everything should start to make sense once we cover the next couple of chapters.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 11: Let us now apply our custom accumulator in the foreach higher order function and print the results to console.

data.foreach(record => {

countByMovie.add((record.location, 1))

})

println(countByMovie.value.toList)

}

}

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Here, we are passing our data through the foreach function, where our custom accumulator named, countByMovie is applied with the add method. We specify the movieId as the field for which the aggregations must be performed. Finally, we can access the result by calling value method on our custom accumulator and convert it to a List.

Step 12: Let us now run the program. We should see the output with count for each movie in a List collection as depicted in the screenshot below.

Please be careful while using the accumulators. If the output generated from the accumulator is a large volume of data, we should not use the accumulators. Instead, we should use the transformations as per requirement. In this case, the result of accumulator is just movies and their counts. It is not a huge data. We have achieved our result without shuffling the data across the network, which is usually the case with transformations.

Task 5 is complete!

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Task 6: Using Broadcast Variables

Let us now look at another type of Distributed Shared Variable called the Broadcast variable. Let us use our movies dataset, which we have been using throughout this course, and evaluate the number of ratings for each movie. We shall be using the movies.csv and ratings.csv files during this task. Moreover, we will broadcast the movies.csv file to look up with the movie Id in ratings.csv file.

Step 1: We will need two files for this lab exercise. Please download and save these files to the IdeaProjects/Spark/chapter_6 folder, if not saved already.

ratings.csv - http://bit.ly/2QmnAH9

This file contains four columns: userId, movieID, rating and timestamp.

movies.csv - http://bit.ly/2EJj0Os

This file contains three columns: movieID, movieName and genre.

Step 2: Open IDE, right-click the training package, which we had created in the previous exercise and hover over New and then click on Scala Class. When prompted, enter ratingsByMovies as the name, click on the dropdown for Kind and select Object. We will need the following imports for our Spark App.

import org.apache.spark.SparkContext

import scala.io.Source

The first import is required to create the SparkContext object, whereas, the second import is a Scala-specific import, which helps us to read the movies.csv file.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Let us now define a function, which would load the movie names to a Map object.

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

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

val data = Source.fromFile("chapter_6/movies.csv").getLines()

for (record <- data) {

val fields = record.split(',')

if (fields.length > 1) {

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

}

}

movieNames

}

We are defining a function called loadMovieNames, which does not take any arguments and returns a Map object, which maps Int to String. Subsequently, we declare a movieNames variable of type Map which maps Int to String and initialize it as an empty map.

Next, we load the data from our file by using the Source.fromFile method and call getlines method to get each line based on /n character. Next, we iterate through each record in our input data by using the ‘for’ comprehension, and split each field based on comma, as we are aware that our fields are delimited by a comma. Next, we check if each record has two fields and finally map movie Id with the movie name, by adding the fields to movieNames Map object. Finally, return the Map object as required by our function.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 3: Let us now write our main function, create a SparkContext object and declare our broadcast variable.

def main(args: Array[String]): Unit = {

val sc = new SparkContext("local[*]", "Ratings By movie ")

val broadNames = sc.broadcast(loadMovieNames)

We can create a broadcast variable by calling the broadcast method on our SparkContext object. Subsequently, we pass our loadMovieNames function as a parameter to the broadcast method since loadMovieNames function returns a Map object. In this manner, we have the Map object broadcasted to all the nodes of the cluster.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 4: Let us now load our ratings.csv file and create an RDD. Next, we split the records based on comma and extract the movieId field. Subsequently, we use the map function to create a pairedRDD of movie Id and its count by 1, so that we can count the number of times each movie is rated in the next step.

val data = sc.textFile("chapter_6/ratings.csv")

val records = data.map(x => (x.split(",")(1).toInt, 1))

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Step 5: Now that we have a pairedRDD, we can use the reduceByKey function to count the number of times each movie is rated as shown below.

val count = records.reduceByKey((x,y) => x + y)

Let us now sort the count, i.e., second field in descending order, so that the highest number of rated movies are on the top of our result.

val sorted = count.sortBy(-_._2)

Step 6: Finally, let us use the broadcast variable to explore the name of movie based on its movie Id. We use the ‘-‘ symbol to sort in the descending order.

val sortedMoviesWithNames = sorted.map(x => (broadNames.value(x._1), x._2))

We are using the map of higher order function to look up the value of the second field from the broadcast variable, which is the movie name, and the second field in our sorted RDD, which is the count. We use the value method to get the value in the broadcast variable. The compiler searches the movie Id with its movie name, and provides us with the name of the movie as the first field in the result and count as the second field.

Accum Method  >> Record Parser  >>  Implem Counters

Accum V2.

Custom Accum

Broadcast Variables

Let us print out the result to the console.

sortedMoviesWithNames.collect.foreach(println)

Step 7: Let us now run our program and check the results. We should see the output as shown below.

Task 6 is complete!

SUMMARY

In this chapter, we have learned the Distributed Shared Variables in detail. Moreover, we explored various types of Accumulators including, Accumulators V2, Custom Accumulators, and Broadcast Variables. Finally, we learned about Scala’s Monadic collections for error handling.

In the labs, we learned to program and utilize various Accumulators, such as Accumulators V2, Custom Accumulators, and Broadcast Variables.

REFERENCES

● https://spark.apache.org/

● https://spark.apache.org/docs/latest/api/java/org/apache/spark/util/AccumulatorV2.html

● https://dzone.com/articles/simplifying-monads-in-scala

CHAPTER 7:

SPARK SQL

Theory

Until this point, we have been working with RDDs in this book, which are the low-level APIs and basic data structures of Spark. Let us now explore Spark SQL and understand what it is and why it is used in Spark.

Types of Data

Before we dig into Spark SQL, let us first look at various types of Big Data available. Primarily, it can be classified in three types.

Structured Data

Structured data is the type of data, which has a schema associated with it. All the attributes, such as data type and data size  etc. related to the data are already available. Usually, this data is stored in tables with table names and their corresponding column headers. An example of unstructured data is Relational Database Management System (RDBMS) which includes Oracle, MySQL, IBM DB2, Microsoft Access etc.

Semi Structured Data

In case of semi-structured data, the associated schema is not completely available. This type of data does not qualify to be stored in a database as there is no data type for the fields, but it will still make sense with appropriate delimiters, which differentiate between the lines and fields. A few examples of semi-structured data is CSV, TSV, XML, and JSON  etc.

Unstructured Data

Unstructured data is the type of data, which does not have any kind of schema associated with it. The examples of unstructured data are presentations, word processing documents, audio, video files, and webpages  etc. Since there is no associated structure, it gets tough to analyze such data. Many advanced technologies such as Artificial Intelligence (AI) and Machine Learning (ML) are used to analyze the unstructured data.

Spark SQL treats both the structured and semi-structured data as the structured data. Hence, this is different behavior when compared to the traditional database management systems like MySQL and Postgress etc.

What is Spark SQL?

Spark SQL is a library or module of Spark, which provides SQL style programming to process the structured data. Spark SQL runs on top of Spark by wrapping the entire Spark core APIs into a high-level abstraction. Moreover, Spark SQL provides optimizations to run the jobs faster which lacks in Spark core, thereby 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 and faster with fewer 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 due to its provision for the developers with easy-to-use APIs with support to 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, and ORC etc. are supported by Spark SQL. Moreover, it also supports almost all the relational databases with JDBC connectivity, which include MySQL, Postgress, and Oracle to name a few. In addition, NoSQL data stores, such as Hbase, Cassandra, and EasticSearch are also supported with Spark SQL.

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

● Spark SQL efficiently processes the structured data by advanced optimization techniques, such as cost based optimizer, in-memory columnar caching, and 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 the structured data. This is similar to the textFile, binaryFile, Sequence File APIs in Spark core (RDD). Instead of numerous APIs for different formats of data, we have Data Source API, which can load and store structured data. Data Source API has built-in support for JDBC, Hive, Avro, JSON, and Parquet, etc. Data Source API can automatically infer schema without the user explicitly mentioning the schema. Moreover, we can also specify the schema using Data Source API.

Data Frames

Data Frames are similar to the 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 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 a distributed manner.

SQL Interpreter & Optimizer

The queries on Data Frames are run in SQL, which is a high-level language. Therefore, we need an SQL Interpreter or an SQL Parser, which will interpret or parse our SQL queries.  The Optimizer in Spark SQL is called 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.

Please check the link in References section to learn more about SQL’s  Catalyst Optimizer.

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. Therefore, instead of running queries from BI tools via  Hive Thrift server as Map Reduce jobs, we can utilize the Spark Thrift Server and use Spark’s features. Since Spark is faster than Hadoop Map Reduce, we can have our queries processes faster in an efficient manner. The BI tools can be connected with Spark by using the JDBC or ODBC drivers. For example, Tableau can be connected to Spark SQL by using the ODBC driver and then run the queries from Tableau on Spark.

Tungsten

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

That is all regarding the theory for this chapter. Let us now proceed to our lab exercises and get our hands dirty with Spark SQL.

AIM

The aim of the following lab exercises is to start writing Spark SQL code in IntelliJ IDEA to learn about Data Frames.

The labs for this chapter include the following exercises:

● Creating Data Frame by using Data Source API

● Creating Data Frame from an RDD

● Creating Data Frame using StructType

● Querying data by using Spark SQL

● Joins by using Spark SQL

● Operations by using DataFrame API

We need the following packages to perform the lab exercises:

● Java Development Kit (JDK)

● Scala

● Spark

LAB EXERCISE 6: SPARK SQL – HANDS ON

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

1. Creating Data Frame by using Data Source API

2. Creating Data Frame from an RDD

3. Creating Data Frame by using StructType

4. Querying Data by using Spark SQL

5. Joins by using Spark SQL

6. Operations by using DataFrame API

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Let us start this exercise by loading a file using Data Source API.

Task 1: Creating Data Frame using Data Source API

Step 1: Download the us-500.csv file from the URL below. This file contains twelve columns: first_name, last_name, company_name, address, city, country, state, zip, phone1, phone2, email, and web.

us-500.csv - http://bit.ly/2LmgDW2

Please save this file in the IdeaProjects/Spark/chapter_7 folder. (Please create a folder named chapter_7 in the home folder.)

Step 2: Open IDE, right-click the training package that you have created in the previous exercise and hover over New and then click on Scala Class. When prompted, enter users as the name, click on the dropdown for Kind, and select Object. Enter the import as shown below.

import org.apache.spark.sql.SparkSession

Since we are using SparkSession object to initiate our Spark Session, we need not import the SparkContext object as we did in the previous exercises. The SparkContext object is wrapped within the SparkSession object in Spark 2.x version.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 3: Let us now write the main function for our program and create a SparkSession object as shown below.

def main(args: Array[String]): Unit = {

val spark = SparkSession

.builder()

.appName("Users")

.master("local[*]")

.getOrCreate()

We are calling the builder method on SparkSession object to build a Spark Session. Next, the appName and master methods are utilized to specify the name of our app and mode of execution (local or cluster), as we did while creating a SparkContext object. Finally, we use the getOrCreate method to get a SparkSession if there is one already or create a new SparkSession if it does not exist.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 4: Let us now load our file using the code as shown below.

val users = spark.read

.format("csv")

.options(Map("inferSchema" -> "true", "header" -> "true"))

.load("chapter_7/us-500.csv")

We call the read method on our SparkSession object spark, which we created in the previous step, and specify the csv as format for our file using the format method. Next, we use a Map object to specify that our input file contains a header, and also ask Spark SQL to infer schema. Since Map contains key-value pairs, the keys are to specify properties and the values are true for both. Finally, we specify the path of our file by using the load method. Please note that we use options method (plural) to specify more than one option by using a Map object. If we only want to specify one option, we should use option method (singular).

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 5: We now have successfully created a DataFrame named users. Let us now print it to the console along with the schema.

users.printSchema()

users.show()

}

}

We call the printSchema method to display the inferred schema and show method to display our DataFrame. Please note that when we use show method, only first 20 records in the DataFrame are shown. We can pass an integer for a number of records in the show method. For example, to show 40 records we can use something like this users.show(40)

The show method is an action, and so this is the point where the DAG is actually executed.

Step 6: Let us execute this program and check the output. We should see the schema as shown below.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

As we can see, the schema has been correctly discovered by Spark for every column in the DataFrame. Please note that if a column has values of more than one data type, Spark will infer it as String.

The output of DataFrame users is as shown below.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Ops. Using DF API

As we can observe from the screenshot above, the header is displayed correctly along with the records.

Step 7: We can also select only one or multiple columns from the DataFrame and have it shown by using the code below.

users.select("last_name").show()

The output is as shown in the screenshot below.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Selecting multiple columns.

users.select("first_name”, “last_name").show()

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

We simply call the select method on the users DataFrame and pass the required columns as arguments. Afterwards, we call the show method as usual.

We can load data to Spark using this same method for any format like JSON, Parqet, ORC, and Avro, etc. Please check the link in the References section for more info.

Task 1 is complete!

Task 2: Creating DataFrame from an RDD

We can also create a DataFrame from an RDD. Let us see how to achieve this.

Step 1: Download the mlb_players.csv file from the URL below. This file contains six columns: name, team, position, height, weight, and age.

mlb_players.csv - http://bit.ly/2JhzVJj

Please save this file in the IdeaProjects/Spark/chapter_7 folder.

Step 2: Open IDE, right-click the training package that you have created in the previous exercise and hover over New and then click on Scala Class. When prompted, enter rddToDf as the name, click on the dropdown for Kind, and select Object. Enter the import as shown below.

import org.apache.spark.sql.SparkSession

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 3: Let us now create a case class, so that we can define schema to our DataFrame. The names, which we specify for the attributes of case class object, will get mapped as column names for our DataFrame. This will make sense when we execute the program.

case class Players(player_name: String, team: String, position: String, height: Int, weight: Int, age: Double)

Step 4: Let us now write the main function for our program and create a SparkSession object as depicted below.

def main(args: Array[String]): Unit = {

val spark = SparkSession

.builder()

.appName("RDD to DataFrame")

.master("local[*]")

.getOrCreate()

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 5: Since our aim is to convert an RDD to a DataFrame, we must use the textFile API in the SparkContext object to read the file and create an RDD.

val input = ss.sparkContext.textFile("chapter_7/mlb_players.csv")

We now have an RDD created. However, the file contains a header with column names. We must first remove the header. We can achieve that by calling the first method on our RDD and then remove it by using the filter method as shown below.

val header = input.first()

val records = input.filter(x => x != header)

The first line of code takes the first record from the RDD, which in our case is the header or column names; and subsequently, we simply filter out the header from input RDD. Now, we just have the records RDD without the column names.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 6: The next step is to split the fields separated by commas, so that we can assign each individual field to its appropriate case class field.

val fields = records.map(record => record.split(","))

Now that we can access individual fields by their position, let us assign them to the case class Players, using the map function as shown here:

val structRecords = fields.map(field => Players(field(0).trim, field(1).trim, field(2).trim, field(3).trim.toInt, field(4).trim.toInt, field(5).trim.toDouble))

We call the trim method on all the fields to remove leading and trailing white spaces, and cast height, weight and age fields to Int, Int and Double, respectively.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 7: We now have our data in structured columns with named records. We can now simply convert it to a DataFrame by using toDF method.

However, before we can use the toDF method, we need to import the implicits as shown below.

import ss.implicits._

val recordsDf = structRecords.toDF()

We now have our DataFrame recordsDf created from RDD.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 8: Let us now call the show method on our DataFrame and execute the program.

recordsDf.show()

}

}

The output is as shown in the screenshot below.

Task 2 is complete!

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Task 3: Creating Data Frame using StructType

In the previous task, we have created a DataFrame from an RDD. We have used a case class and toDF method to achieve the same. However, there are some limitations to using the case class method. The case class cannot have more than 22 arguments. If our data has more than 22 fields, it becomes hard to create a DataFrame from RDD by using case class and toDF method.

To overcome this limitation, we have a createDataFrame method, which takes an RDD and schema as parameters to create a DataFrame. Let us create a DataFrame by using the createDataFrame method.

We shall be using the same input file mlb_players.csv for this task as well. However, you are highly encouraged to use another input file and play around.

Step 1: Open IDE, right-click the training package that you have created in previous exercise and hover over New and then click on Scala Class. When prompted, enter createDf as the name, click on the dropdown for Kind, and select Object. Enter the import as shown below.

import org.apache.spark.sql.SparkSession

import org.apache.spark.sql.Row

import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType}

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 2: Let us now write the main function and create SparkSession as we already did in the previous tasks.

def main(args: Array[String]): Unit = {

val ss = SparkSession

.builder()

.appName("Rdd to DataFrame")

.master("local[*]")

.getOrCreate()

Step 3: The next step is similar to what we have done in the previous task. We load the input file using the textFile API, extract the header and filter it out by using the filter method. Next, we split the fields based on comma delimiter.

val input = ss.sparkContext.textFile("chapter_7/mlb_players.csv")

val header = input.first()

val records = input.filter(x => x != header)

val fields = records.map(record => record.split(","))

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 4: Now, instead of mapping the fields to a case class as in the previous task, we map the fields to a Row object. This Row object is an ordered collection of fields, which can be accessed by index or position. It is similar to a row in a table.

val structRecords = fields.map(field => Row(field(0).trim, field(1).trim, field(2).trim, field(3).trim.toInt, field(4).trim.toInt, field(5).trim.toDouble))

We now have our fields as rows. All the fields are assigned and casted as we did in the previous task.

Step 5: Now that we have rows, let us create a schema. We can create schema by using the instance of StructType object. The StructType object contains StructField objects, which take parameters as name of the column, type of the column and an optional Boolean specifying if the column contains null values. In addition, the data type must be defined as StringType, IntegerType, and DoubleType etc.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

val schema = StructType(List(

StructField("player_name", StringType, false),

StructField("team", StringType, false),

StructField("position", StringType, false),

StructField("height", IntegerType, false),

StructField("weight", IntegerType, false),

StructField("age", DoubleType, false)

))

We have specified the StructFields as a List inside the StructType object.

Step 6: Finally, we can use our RDD, which is a structRecords and schema as parameters for the createDataFrame method to create a DataFrame.

val recordsDf = ss.sqlContext.createDataFrame(structRecords, schema)

Since createDataFrame is a method of sqlContext object, we call sqlContect on our SparkSession object and then call createDataFrame method.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Let us use the show method to check the created DataFrame by using createDataFrame method.

recordsDf.show()

}

}

Step 7: Let us finally execute our program and check the output. The output is as shown in the screenshot below.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

We have successfully created our DataFrame by using the createDataFrame method.

Task 3 is complete!

Task 4: Querying data using Spark SQL

With Spark SQL, we can process data in two ways. The first way is to use the Spark SQL data manipulation language, which are essentially the traditional SQL queries in RDBMS. We can simply perform operations by using SQL queries and obtain the result. The second way is to use the DataFrame API, which is the DataFrame Data Manipulation language.

We can use any of these methods to process our data based on our levels of comfort. Both the APIs are optimized and are efficient. Let us first use the SQL queries to process data in this task and look at few operations we can perform.

Step 1: We shall be using the file us-500.csv for this task as an input source. Please create a new object and name it sqlQueries. But first, perform all the steps you performed in Task 1. Your program should look something like the screenshot shown below.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 2: Let us first assign a view for our DataFrame, so that we can execute queries against it. In simple words, we are just creating a table for our DataFrame, so that we can reference it while we run SQL queries against it.

users.createOrReplaceTempView("users")

We are using the createOrReplaceTempView method to create a temporary table named users if it does not exist, or replace a view if it already exists with the same name. This temporary table is available until the SparkSession is active; once it finishes, the table will not be available anymore. Hence, the name is given as the temp view. We can also persist the table using saveAsTable method.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 3: Let us now run some queries. First, let us execute a basic query to select all the users from the table who belong to the state of Florida.

val foridaUsers = spark.sql("SELECT * FROM users WHERE  state = \”FL\”")

We use the sql method in our SparkSession object, which is spark, and enter the following query. We have simply entered a query to select all the records from our users table who belong to state of Florida (or FL). Since the values of State are in the form of String, we have to enclose them in double quotes and use the escape character ‘\’.

Next, we can simply call the show method on floridaUsers DataFrame to check the results.

results.floridaUsers.show()

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Run the program and you should have the result as shown in the screenshot below with the top 20 users who belong to State ‘FL’.

You can also use the collect, foreach and print methods to print all the records as we used to in the previous exercises.

Step 4: Let us now run a query to check the count of total users who belong to the state of “New Jersey.”

val totalUsersNJ = spark.sql("SELECT count(*) AS NJ_Count FROM users WHERE state = \”NJ\”")

totalUsersNJ.show()

In the query above, we are simply using the count function with a WHERE clause to get the count of users who belong to the state of NJ. We utilize the AS clause to name the column as depicted in the screenshot below. Subsequently, we use the show method to display the results.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

The output should be as shown in the screenshot below.

Step 5: Let us now look at the count of users by state and arrange them in the descending order of their count.

Note: You may comment on the previous queries to avoid processing them again.

val userCountByState = spark.sql("SELECT state, count(*) AS count FROM users GROUP BY state ORDER BY count DESC)

totalUsersNJ.show()

In the query above, we have utilized GROUP BY query to group by state and then the ORDER BY query to sort the count in the descending order.

The result is as shown in the screenshot below.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 6: Instead of using the show method, let us now see how we can save this file. We may comment out or remove the line with the show method from the previous step.

Before we save the file, there is one important thing we need to know about how SparkSQL creates the tasks. For the second stage of DAG, the number of tasks created to process the data are 200 by default. However, in Spark core, the number of tasks in second stage is always equal to the number of tasks in the first stage. For example, consider we have two stages; stage 0 and stage 1. For stage 0, the number of tasks created will be equal to the number of input splits. Next, in Spark core, the number of tasks in stage 1 are equal to the number of tasks in stage 0. However, in SparkSQL, the number of tasks in stage 1 are 200 by default.

The 200 tasks in SparkSQL is a good starting point if there are large volumes of data to process. We can configure the number for optimization if required. However, our file is just a sample of small data and there is no requirement of 200 tasks to be created. Therefore, using the following property, we can set the total number of tasks as 1. If we run with the default value of 200 and save the file, there will be multiple partitions of output for small datasets. You are free to check how the output looks like without setting this property.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

spark.conf.set("spark.sql.shuffle.partitions", "1")

Since this is a configuration, we are calling conf on our SparkSession object and setting the property to utilize only one task by using the set method. This will only create one task and we are left with only one output file.

Let us now save the file by using the code below.

userCountByState.write

.format("csv")

.save("chapter_7/output")

Similar to reading the file using read and load methods, we use write and save methods to save the file-to-file system.

After we run the program, the process should finish with an exit code of 0 which indicates the successful processing and the output should be available in the file system with only one file in the saved location.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Open the file and you should have the values as shown below.

This way, we can perform any operations by using the SQL data manipulation language.

Task 4 is complete!

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Task 5: Joins using Spark SQL

Let us now use Spark SQL to join two DataFrames.

Step 1: Download the ratings.csv file from the URL below. This file contains four columns: userId, movieID, rating, and timestamp.

ratings-head.csv - http://bit.ly/2FPdhHE

Please save this file in the IdeaProjects/Spark/chapter_7 folder if not saved already in the previous tasks.

Step 2: Download the movies.csv file from the URL below. This file contains three columns: movieID, movieName, and genre.

movies-head.csv - http://bit.ly/2RTg72N

Please save this file in IdeaProjects/Spark/chapter_7 folder.

We shall join these datasets based on the movieID.

Step 3: Create a new Scala object and name it sqlJoins, and then import the following:

import org.apache.spark.sql.SparkSession

Subsequently, we can write the main function for our program and create a SparkSession object as shown below.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

def main(args: Array[String]): Unit = {

val spark = SparkSession

.builder()

.appName("SQL Joins")

.master("local[*]")

.getOrCreate()

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 4: Let us now read both the files as shown below.

val movies = spark.read

.format("csv")

.options(Map("inferSchema" -> "true", "header" -> "true"))

.load("chapter_7/movies-head.csv")

val ratings = spark.read

.format("csv")

.options(Map("inferSchema" -> "true", "header" -> "true"))

.load("chapter_7/ratings-head.csv")

We now have two DataFrames for each of our input files.

Step 4: Now that we have our DataFrames, let us create a temp view, so that we can run our join query against them.

movies.createOrReplaceTempView("movies")

ratings.createOrReplaceTempView("ratings")

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 5: We now have our views. All we need to do now is to perform the join. We can join this by using the SQL query as shown below.

val joinedDf = spark.sql("SELECT * FROM movies JOIN ratings ON movies.movieId = ratings.movieId”

Finally, let us call the show method on our joinedDf DataFrame and run the program.

joinedDf.show()

We can observe the joined table as shown in the screenshot below:

Task 5 is complete!

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Task 6: Operations using DataFrame API

Step 1: The initial few steps are similar to what we have performed in the previous tasks. Create a new Scala object and name it dfOps. Specify the required imports and create a SparkSession object as in the previous tasks. Finally load the us-500.csv file.

Our program at this point of time should look like the one in the screenshot below.

Step 2: Now that we have the DataFrame created, let us perform an operation to select all the users from Florida by using the DataFrame API.

val foridaUsers = users.select(“*”).where(“state = \”FL\”")

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

This is similar to the SQL query that we had performed in the task earlier. We have methods here, which look more like programming style. In the code above, we have used the select method to select all the columns of our DataFrame and then where method to specify our condition.

Please note that we need not create a temp view as we did earlier. It is only required when we are working with the SQL queries.

Let us now call the show method on our DataFrame to view the results on the console.

floridaUsers.show()

The output should look something like the table as shown in the screenshot.

Step 3: Let us now run a query to check the count of total users who belong to the state of “New York.”

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

val nyUserCount = users.groupBy("state")

.agg(("state", "count"))

.where("state = \"NY\"")

This is a bit different as compared to what we performed in the SQL query. In the code above, we are first grouping by state and then applying the agg method. The agg method takes the column as the first parameter and then the type of aggregation as the second parameter. We specify the second parameter as count since we want to count the number of users from the state of New York. Subsequently, we specify the condition using the where method.

Let us now call the show method on our DataFrame to view the results.

nyUserCount.show()

The output is as shown in the screenshot below.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Step 4: Let us now write some code to get the count for all the users for each state. First, we need to import the implicits as shown below.

import spark.implicits._

val userCountByState = users.groupBy("state")

.agg(("state", "count"))

.orderBy($"count(state".desc)

As we can observe in the query above, we use the orderBy method to order the result by the count of state in a descending order.

Let us call the show method.

userCountByState.show()

The output can be observed in the screenshot below.

Creating DF  >> DF for RDD  >>  DF StructTy

Spark SQL.

Spark SQL Joins

Ops Using DF API

Task 6 is complete!

SUMMARY

In this chapter, we have learned the basics of Spark SQL. Moreover, we have also studied the Spark SQL, its applications, and architecture in detail.

In the labs, we have had our hands on creating a DataFrame, converting an RDD to DataFrame by using the toDF method and StructType. Subsequently, we used the SQL queries and the DataFrame API to process data.

REFERENCES

● https://spark.apache.org/

● https://spark.apache.org/docs/latest/sql-data-sources.html

● https://spark.apache.org/docs/latest/api/sql/index.html

CHAPTER 8:

DATASETS

Theory

In the previous chapters, we worked with RDDs and DataFrames, their applications, and architectures. In this chapter, let us look at the Datasets API, which is the most advanced API in Spark. We shall comprehensively learn datasets and their applications. To understand datasets better, let us first understand the difference between an RDD and a DataFrame.

RDD vs. DataFrame

RDD is a low-level API.

DataFrame is a high-level API.

RDDs are written in functional style of programming.

DataFrames are more of a relational style of programming.

RDDs are type safe, which guarantees that the variable of a particular data type declared holds the exact same type of data within it. Type checking happens at the compile time and not at the runtime. For example, if we declare a variable as  Int and assign a String, an error will be shown at compile time itself.

DataFrames are not type safe, where type error is only shown at runtime and not at the compile time. It is possible to declare an Int and assign a String, but this error is only detected at runtime.

RDDs are slower as compared to  DataFrames. There is no optimizer available for RDDs.

DataFrames are faster and efficient, as they are optimized with Catalyst optimizer and Tungsten.

RDDs process structured as well as unstructured data.

DataFrames on the other hand only process structured and semi-structured data.

RDDs cannot infer schema for the ingested data, as they require the users to specify the schema in an explicit manner.

DataFrames have the ability to infer schema automatically for the ingested data. The user can also explicitly specify the schema.

As we can observe from the comparison above, there are clearly some advantages and disadvantages in case of both the APIs. Nonetheless, it will be great if we could somehow take the advantages and discard the disadvantages of each. Well, it turns out that we could.

What are Datasets?

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

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

Why Datasets?

The following are the advantages of using datasets:

● Datasets are the combination of RDDs and DataFrames. They can help us to code similar to what we would do with RDDs and process them through the performance optimization engines.

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

● They assure 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 by using the Catalyst optimizer and Tungsten similar to that of DataFrames. This ensures very fast and efficient processing of data.

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

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

● They can be converted from RDD and DataFrames.

We shall be working with datasets API later in the lab exercises.

That is all in theory for this chapter. Let us now proceed to our lab exercises.

AIM

The aim of the following lab exercises is to start writing Spark SQL code in IntelliJ IDEA to learn about datasets and built-in functions.

The labs for this chapter include the following exercises:

● Creating datasets using DataSource API

● Creating dataset from an RDD

● Aggregate and Collection functions

● Date/Time functions

● Math and String functions

● Window functions

We need the following packages to perform the lab exercises:

● Java Development Kit (JDK)

● Scala

● Spark

LAB EXERCISE 7: DATASETS & FUNCTIONS

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

1. Creating datasets using DataSource API

2. Creating datasets from an RDD

3. Aggregate and Collection functions

4. Date/Time functions

5. Math and String functions

6. Window functions

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Task 1: Creating Dataset using Data Source API

Creating a dataset is similar to that of a DataFrame with some minor changes.

Step 1: Download the ratings.csv file from the URL below. This file contains four columns: userId, movieID, rating, and timestamp.

ratings-head.csv - http://bit.ly/2FPdhHE

Please save this file in the IdeaProjects/Spark/chapter_8 by creating a new folder.

Step 2: Open IDE, right-click the training package that you have created in previous exercise and hover over New and then click on Scala Class. When prompted, enter createDS as the name, click on the dropdown menu for Kind, and select Object. Enter the import as shown below.

import org.apache.spark.sql.SparkSession

Next, we need to write a case class so that we can specify the schema for our fields. This case class creates a dataset and differentiates from DataFrame. While loading the file, we simply refer to this case class object to create a dataset.

private case class Movies(userId: Int, movieId: Int, rating: Double, timestamp: String)

We have created a case class and named it Movies by specifying the fields and its types.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

If there is an error saying that the case class Movies is already defined, change the name of class case to anything else and refer to it accordingly throughout the program.

Step 3: Now, write the main function and create the SparkSession object as shown below.

def main(args: Array[String]): Unit = {

val spark = SparkSession

.builder()

.appName("Creating a Dataset")

.master("local[*]")

.getOrCreate()

Step 4: Before we load the file by using the DataSource API, we need to import the implicits, which is required to create a dataset.

import spark.implicits._

This import is available in our SparkSession object. Hence, we refer it to the SparkSession object (spark) which we had created in the previous step.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Now, we load the file as we usually do while creating the DataFrame except for the as method at the end. The as method refers to the case class object Movies creating a dataset object.

val movies = spark

.read

.format("csv")

.options(Map("header" -> "true", "inferSchema" -> "true"))

.load("chapter_8/ratings-head.csv")

.as[Movies]

.cache()

We use the cache method to cache our dataset in memory so that it is created every time an action is called. We now have our dataset created.

Step 5: We can now use the show and printSchema methods to check the first 20 records in our dataset and the schema associated with it.

movies.printSchema()

movies.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

The following schema is shown when we execute the program.

The result can be observed below:

Step 6: Let us now perform some basic operations with our dataset. The following code below is used to find the total count of each rating.

val ratingCount = movies.groupBy("rating").count()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

We simply use the groupBy function to group all the ratings and then count the number of ratings by using the count function.

Next, let us call the show method to display the result on the console.

ratingCount.show()

Once the program has finished running, the following results are displayed at the console:

However, this was also achieved using DataFrames. Then how are the datasets special? Well, we can also write functional style programming with datasets.

Step 7: Let us now write some functional style programming by using our dataset. First, we shall extract the two columns userId and rating from the movies dataset by using the map function.

val users = movies.map(x => (x.userId, x.rating))

As we can observe, we have used functional style programming the code above which we cannot use with DataFrames. In addition, we are able to refer to the column by its name.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Next, let us convert our dataset to RDD and write some code to find out the average rating for each user.

val count = users.rdd.mapValues(x => (x,1))

.reduceByKey((x,y) => (x._1 + y._1, x._2 + y._2))

.mapValues(x => x._1/x._2)

.sortByKey(false)

We have simply used the rdd method to convert our dataset to RDD. The rest of the code is the familiar functional programming style.

Let us now use the collect method just like in the previous exercises and print out the result to the console.

count.collect.foreach(println)

}

}

The output should be shown as below.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Task 1 is complete!

Task 2: Creating Dataset from an RDD

As we had done in the previous lab exercise to create a DataFrame from an RDD, we shall also look at creating a dataset from an RDD.

Step 1: Download the mlb_players.csv file from the URL below. This file contains six columns: name, team, position, height, weight, age.

mlb_players.csv - http://bit.ly/2JhzVJj

Please save this file in the IdeaProjects/Spark/chapter_8 folder.

Step 2: Open IDE, right-click the training package that you have created in the previous exercise and hover over New and then click on Scala Class. When prompted, enter rddToDs as the name, click on the dropdown menu for Kind, and select Object. Enter the import as shown below.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

import org.apache.spark.sql.SparkSession

Step 3: Let us now create a case class so that we can define a schema to our dataset as we had done with DataFrame in the previous exercise. The names that we specify for the attributes of the case-class object will be mapped as column names for our dataset.

case class Players(player_name: String, team: String, position: String, height: Int, weight: Int, age: Double)

Step 4: Let us now write the main function for our program and create a SparkSession object as shown below.

def main(args: Array[String]): Unit = {

val spark = SparkSession

.builder()

.appName("RDD to Dataset")

.master("local[*]")

.getOrCreate()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 5: Since our aim is to convert an RDD to a dataset, we must use the textFile API in the SparkContext object to read the file and create an RDD.

val input = ss.sparkContext.textFile("chapter_8/mlb_players.csv")

We now have an RDD created. However, the file contains a header with column names. We must first remove the header. Instead of using the first method followed by the filter method, let us remove the header by using an efficient approach.

val removeHeader = input.mapPartitionsWithIndex((index, itr) => {

if (index == 0) itr.drop(1) else itr

})

In the previous approach to remove the header, we were using the filter method to filter out the first record from the RDD. This approach works well when there is small data. However, Spark is not developed for small data. Instead, we will be working on a large volume of data in a real-time environment.

Consider a scenario where we have a billion records in our RDD. When we use the filter function, the filter condition is tested for each record of RDD. Hence, just to filter out one record, we end up testing the condition for all the billion records. To overcome this, we can use the mapPartitionWithIndex function. This higher-order function provides us with an entire partition of RDD with index, unlike the map function, which provides us with each record. Subsequently, we can pass our logic, which is to drop the first record in the partition if the partition index is 0, else return the partition as is. Therefore, the condition inside if statement is only verified per partition basis instead of per record. This condition is only tested based on the number of partitions and not on the number of records, making it efficient.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 6: The next step is similar to what we had done in the previous exercise while creating a DataFrame from RDD. We split the fields based on a comma, so that we can assign each individual field to its appropriate case class field.

val fields = removeHeader.map(record => record.split(","))

Now that we can access individual fields by their position, let us assign them to the case class Players, using the map function as shown here.

val structRecords = fields.map(field => Players(field(0).trim, field(1).trim, field(2).trim, field(3).trim.toInt, field(4).trim.toInt, field(5).trim.toDouble))

We call the trim method on all the fields to remove the leading and trailing white spaces, and cast height, weight and age fields to the types Int, Int and Double respectively.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 7: We now have our data in the structured columns with named records. We can now simply convert it to a dataset by using the toDS method.

Nevertheless, before we can use the toDS method, we need to import the implicits as shown below.

import ss.implicits._

val recordsDs = structRecords.toDS()

We now have our dataset recordsDs created from RDD.

Step 8: Let us now call the show method on our dataset and execute the program.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

recordsDs.show()

}

}

The output is as shown in the screenshot below.

Therefore, the only difference between creating a DataFrame and a dataset is the method, which we call at the end. We use the toDF method to create a DataFrame and the toDS method to create a dataset.

We can also utilize the programmatical schema to create a dataset as we had done with the DataFrame in Task 3 of previous exercise. Please try it out and create a dataset by programmatically creating a schema.

Task 2 is complete!

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Task 3: Aggregate and Collection Functions

Before we look at the user-defined functions, let us look at few of the built-in functions, which are available in Spark in the next two tasks.

Aggregate Functions

Step 1: Please perform the steps 1 – 4 from Task 1 in this exercise. Make sure that you name the Scala object as builtInFunctions instead of createDS. Moreover, make sure you rename the case class to Rating so that it does not conflict with classes from other objects. We should be having a screen as shown below.

Step 2: Let us first use the built-in aggregation functions. In the following piece of code, we are performing multiple aggregations at once.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Before we use the functions, we need to include the following import as shown below.

import org.apache.spark.sql.functions._

Next, we can simply use the select method and perform the multiple aggregations at once.

val agg = movies.select(

avg("rating")

, min("userId")

, max("movieId")

, sum("userId")

, mean("rating")

)

In this case, we have used multiple aggregations such as avg, min, and max etc. on various columns. As the names of the functions suggest, avg computes the average, whereas, the min and max aggregations compute the lowest and the highest values in the column. Moreover, the sum and mean aggregations compute the sum and mean of all the values in the columns, respectively.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 3: Let us call the show method and run the program.

agg.show()

The following result should be shown.

Step 4: As we can observe from the screenshot above, the column names are not very meaningful. Therefore, we can assign our own column names by using the as method.

val aggAlias = movies.select(

avg("rating").as("avgRating")

, min("userId").as("lowestUserId")

, max("movieId").as("highestMovieId")

, sum("userId").as("sumOfUserId")

, mean("rating").as("meanRating")

)

Here, we are doing is assigning an alias (column name) to each field using the as method.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 5: Let us call the show method and run the program.

aggAlias.show()

The following result should be shown.

As we can observe from the screenshot above, the column names appear as we have specified.

Step 6: Let us now use the groupBy method and perform the aggregations.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

val byUser = movies.groupBy("userId")

.agg(countDistinct("rating").as("distinctCount")

, sumDistinct("rating").as("distinctSum")

, count("movieId").as("movieCount"))

In the code above, we have used the groupBy method and performed the aggregations over the group. We have used the countDistinct function to count the number of distinct ratings for userId. Similarly, the sumDistinct function is used to sum the distinct ratings only. Finally, the count function, as we know, is used to count all the movie ids.

Step 7: Let us call the show method and run the program.

byUser.show()

The following result should be shown.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Collection Functions

So far, we have been working with IDE and learned how to create objects and execute the programs. However, in this task, we learn about the collection functions. Let us use the Spark shell, to check the output on the fly quickly instead of running the program every time we use a function. We may choose to work with IDE and it is perfectly fine. We will then have to create the SparkSession object and specify the required imports.

Step 1: Open the terminal and fire up the Spark shell. We should be having the Scala prompt as shown in the screenshot below.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 2: We need to import the implicits and functions to be able to work with the functions.

import spark.implicits._

import org.apache.spark.sql.functions._

Let us now declare a Seq collection as shown below.

val num = Seq(Seq(1,2,3), Seq(4, 5, 6), Seq(7,8,9))

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 3: Next, let us convert the collection to a dataset using the toDS method and rename the column as numbers by using the withColumnRenamed method. The default column name when we create a dataset is value. Hence, we change the default column name to numbers.

val numDS = num.toDS().withColumnRenamed("value", "numbers").cache()

Step 4: Now that we have our dataset, let us apply some of the collection functions. First, let us use the array_contains function to check if the collection contains the element we require.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

val contains = numDS.where(array_contains($"numbers", 5))

We use the array_contains function inside the where method to check if the column numbers contains the number 5. This function takes the column name as the first argument and the element as second. Moreover, we can also pass more than one element as the second argument enclosed in a collection as shown in the example below.

val eg = numDS.where(array_contains($"numbers", Array(7,8))

Let us check the result by using the show method.

contains.show()

The result is as depicted in the screenshot below. Since we have used the where method, we can only see the collection which contains the number 5.

Step 4: Let us now use the explode function, which takes each element in a collection and generates a new row.

val exploded = numDS.select($"numbers", explode($"numbers").as("exploded"))

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

In the code above, we have used the select method to select the numbers column, and then the result of using explode function on numbers column as the second column. We have specified the second column name as exploded.

After running the show method, the following result can be observed as shown in the screenshot below:

exploded.show()

As we can observe from the output, each element inside the collection is a new row.

Similarly, we also have the posexplode function, which also provides us with the position of each row.

val posExploded = numDS.select(posexplode($"numbers"))

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

When the posexplode function is used, two different columns are created. One is the column with the exploded values as observed above and the other is with the position of each exploded value.

After running the show method, the following result is shown as in the screenshot.

posExploded.show()

Since we have not used the as method to specify the names of the columns, the default column names pos and col are created for us.

Step 5: Let us now use the size function to check the size of each of the collections in our datasets. It simply returns the size of the collection similar to the length method.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

val sizeDS = numDS.select($"numbers", size($"numbers").as("size"))

We have used the select method to select the numbers column as the first column and size function on numbers column as the second column. Subsequently, we have used the as method to rename the column as size.

After running the show method, the following result is shown as in the screenshot.

sizeDS.show()

The size is shown as 3 since there are three elements in each collection.

Step 6: Let us now utilize the sort_array function to sort the collection in ascending or descending order.

val sorted = numDS.select($"numbers", sort_array($"numbers", false).as("sorted"))

The sort_array function takes the column on which we want to perform the sort on as the first argument and Boolean as second argument. We specify true if we want to sort in the ascending order or false, in case of the descending order.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

After running the show method, the following result is depicted as in the screenshot.

sorted.show()

There are many other functions, which can be applied. Please check the link in references section and try to practice the application of all the functions.

Task 3 is complete!

Task 4: Date/Time Functions

Let us now look at Date/Time functions to manipulate, extract and perform arithmetic operations on date and time. As in the case of collection functions, we shall be using the Spark shell to demonstrate Date/Time functions as well. Please feel free to use IDE if you prefer it over Spark Shell.

Step 1: Let us first create a collection with data as shown below. Please make sure you have the imports from the previous section already available. Otherwise, you will have to import them again if you had closed the Spark Session.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

val dates = Seq(

(1, "Ernesto", "2015-09-24"),

(2, "Lee", "1985-05-16"),

(3, "John", "2012-07-16"),

(4, "Doe", "1914-08-02")

)

Next, let us convert the collection to a dataset using the toDS method and rename the column as shown below by using the withColumnRenamed method. The default column names for dataset are monotonically increasing numbers like _1, _2, _3 etc.

val datesDS = dates.toDS()

.withColumnRenamed("_1", "id")

.withColumnRenamed("_2", "name")

.withColumnRenamed("_3", "date")

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Let us check the schema by using the printSchema method, so that we can compare the data type for date column in the next step.

datesDS.printSchema()

As we can observe from the screenshot above, the date is of type String.

Step 2: Let us cast the date column and convert it to date type by using the cast function as shown below.

val casted = datesDS.select($"id", $"name", $"date".cast("date")).cache()

Let us print the schema to check if we were able to convert the date column successfully from String type to Date type. Moreover, let us also use the show function to view the dataset.

casted.printSchema()

casted.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

As we can observe from the screenshot above, we have successfully casted the date column as date type.

Step 3: Let us now extract the individual attributes from the date object such as day, month, and year etc. We shall be using various functions to add the columns for each function using the withColumn method.

val extracted = casted

.withColumn("year", year($"date"))

.withColumn("month", month($"date"))

.withColumn("dayOfYear", dayofyear($"date"))

.withColumn("quarter", quarter($"date"))

.withColumn("weekOfYear", weekofyear($"date"))

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

We have used the year, month and dayofyear functions to extract the individual attributes from the date column. Furthermore, we have also used the quarter function to get which quarter the date is from and the weekofyear function to get the corresponding week to which the date belongs.

The following output is shown when we utilize the show method.

extracted.show()

Step 4: Let us now use the arithmetic functions to manipulate the date.

val arithmetic = casted

.withColumn("ageInDays", datediff(current_date(), $"date"))

.withColumn("addedDays", date_add($"date", 25))

.withColumn("subtrDays", date_sub($"date", 16))

.withColumn("addedMonths", add_months($"date", 4))

.withColumn("lastDay", last_day($"date"))

.withColumn("nextDay", next_day($"date", "tuesday"))

.withColumn("monthsBetween", months_between(current_date(), $"date", true))

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

● The datediff function is used to calculate the date difference between two dates. Here we have used the current_date method to get the present date and acquire the difference from the date in date column.

● The date_add and date_sub functions are utilized to add and subtract the number of days from the date in date column. The function takes the date column and the number of days as arguments.

● The add_months function is used to add the number of months to the date in date column. The function takes the date column and the number of months as arguments.

● The last_day and next_day functions are used to get the last day of the month and next day of the month (for the day of the week for the current date) in date column respectively. The next_day function takes date column and the day of week as arguments.

● The months_between function is used to get the total number of months between two days. We have used the current date by using current_date function and date column as the arguments.

The following output is observed when we use the show method.

arithmetic.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 5: Next, let us utilize the timestamp functions. Since we have only created a date type in the previous dataset, let us create a timestamp type instead of date type. First, let us create the dataset and rename the columns as shown below.

val timeStamp = spark.createDataset(Seq(

(1, "Ernesto", "2015-09-24 00:01:12"),

(2, "Lee", "1985-05-16 03:04:15"),

(3, "John", "2012-07-16 06:07:18"),

(4, "Doe", "1914-08-02 09:10:20")

))

val timeStampDS = timeStamp

.withColumnRenamed("_1", "id")

.withColumnRenamed("_2", "name")

.withColumnRenamed("_3", "timeStamp")

Let us print the schema, so that we can compare it with the timestamp type in the next step.

timeStampDS.printSchema()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 6: Let us now convert the timestamp that is of String type to timestamp type.

val castedTimeStamp = timeStampDS.select($"id", $"name", $"timeStamp".cast("timestamp")).cache()

Let us now print the schema and the dataset to check the casting.

castedTimeStamp.printSchema()

castedTimeStamp.show()

As we can observe from the screenshot below, we have successfully casted the timestamp column from String type to timestamp type.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 7: Let us now extract the attributes from the timestamp column as we had done for the date column couple of steps ago.

val extractedTs = timeStampDS

.withColumn("second", second($"timeStamp"))

.withColumn("minute", minute($"timeStamp"))

.withColumn("hour", hour($"timeStamp"))

The functions used above are self-explanatory.

The following output is shown when we use the show method.

extractedTs.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 8: Finally, let us use a couple of conversion functions to convert the dates into different formats.

val conversions = timeStampDS

.withColumn("unixTime", unix_timestamp($"timeStamp"))

.withColumn("fromUnix", from_unixtime($"unixTime"))

● The unix_timestamp function is used to convert the timestamp to UNIX timestamp.

● The from_unixtime function is used to convert the UNIX time, which we obtained as shown in the above screenshot.

The following output is shown when we use the show method.

conversions.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Task 4 is complete!

Task 5: Math and String Functions

Math Functions

There are a number of math functions, which can be applied to the columns with numbers. Let us now look at a few of these functions.

Step 1: Let us first create a collection with data as shown below. Please make sure you have the imports from the previous section already imported. We will have to import them again if we have closed the Spark Session.

val numbers = List(5, 4, 9.4, 25, 8, 7.7, 6, 52)

Step 2: Next, let us convert the collection to a dataset by using the toDS method and rename the column as numbers using the withColumnRenamed method. The default column name when we create a dataset is value. Hence, we change the default column name to numbers.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

val numbersDS = numbers.toDS()

.withColumnRenamed("value", "numbers")

.cache()

The dataset should now be created with the renamed column.

Step 3: Let us now perform various math functions on the dataset. All these functions are self-explanatory.

val mathFuncs1 = numbersDS.select(abs($"numbers"), ceil($"numbers"), exp($"numbers"), cos($"numbers"))

● The abs function returns the absolute value of the number.

● The ceil function returns the number of double type greater than or equal to the nearest rounded integer.

● The exp function returns the Euler’s e raised to the power of double value.

● The cos function returns the trigonometric cosine of an angle.

Let us check the result using the show method.

mathFuncs1.show()

The following result is shown.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 4: Let us now use some additional math functions.

val mathFuncs2 = numbersDS.select(factorial($"numbers"), floor($"numbers"), hex($"numbers"), log($"numbers"))

● The factorial function returns the factorial of the number.

● The floor function is opposite to the ceil function which returns the number of double type lesser than or equal to the nearest rounded integer.

● The hex function returns a hex value.

● The log function returns the natural logarithm (base e) of a double value as a parameter.

Let us check the result by using the show method.

mathFuncs2.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

The following result is shown.

Step 5: Let us now use even more math functions.

val mathFuncs3 = numbersDS.select(pow($"numbers", 2), round($"numbers"), sin($"numbers"), tan($"numbers"))

● The pow function returns a number raised to the power of some other number. It takes two arguments: the first argument is the column with numbers and the second argument is the number which power has to be calculated.

● The round function returns the rounded value to the nearest decimal.

● The sin and tan functions return the sine and tangent trigonometric angles, respectively.

Let us check the result by using the show method.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

mathFuncs3.show()

The following result is shown.

Step 6: Let us finally conclude math functions with a couple more of them.

val mathFuncs4 = numbersDS.select(sqrt($"numbers"), log10($"numbers"), $"numbers" + Math.PI)

● The sqrt function returns the square root of the given numbers in the column.

● The log10 function returns the base 10 logarithm of a double value.

● In the third column, we have simply added the value of PI by using the Math.PI expression.

Let us check the results by using the show method.

mathFuncs4.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Many other math functions can be applied. Please check the references section for link to all the math functions.

String Functions

There is a plethora of String functions available in Spark. Let us look at a few of them now.

Step 1: As usual, let us first create the List and create a dataset from it. Please make sure to specify imports again if you had closed the Spark session.

val quote = List("I have no special talent.",

"I am only passionately curious.",

"I have a dream.",

"I came, I saw, I conquered.")

val quoteDS = quote.toDS().cache()

Let us use the show method to display the dataset as shown below.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

quoteDS.show()

Step 2: Let us first use the split method to split the strings by space as below.

val splitted = quoteDS.select(split($"value", " ").as("splits"))

The split method takes two arguments: the first argument is the name of the column and the second method is the pattern to which the string should be split on.

Let us use the show method to display the dataset as shown below.

splitted.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

As we can observe from the screenshot above, the rows have now been split by whitespace.

Step 3: Next, let us use the explode function, which we have learned in this exercise to create a column for each element in the collection.

val exploded = splitted.select(explode($"splits").as("explode"))

Let us use the show method to display the dataset as shown below.

exploded.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Now that we have each word as a row, let us apply some string functions.

The first function is for the evaluation of the length of each word using the length function as shown below.

val strLen = exploded.select($"explode", length($"explode")).as("length")

Let us use the show method to display the dataset as shown below.

strLen.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 4: Let us now use other string functions to convert strings from the lower case to the upper case, and vice versa.

val upCase = quoteDS.select(upper($"value").as("upperCase"))

val lowCase = upCase.select(lower($"upperCase").as("lowerCase"))

Let us utilize the show method to display the datasets as shown below.

upCase.show()

lowCase.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 5: Finally, let us look at substring and trim functions to extract a part of string and trim the whitespaces before and after that string, respectively.

val sub = quoteDS.select(substring($"value", 0, 2).as("firstWord"))

val trimmed = sub.select(trim($"firstWord"))

Let us use the show method to display the datasets as shown below.

sub.show()

trimmed.show()

The substring function takes three arguments. The first is the column name from which the sub string should be extracted. Second and third arguments are provided to specify the start and end positions from which we extract the string.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Here we have extracted the first word and space after it using the substring function. Since our first word is only a letter, we start from 0 position and end at 2nd position which is a whitespace. Next, we utilize the trim function to trim the whitespaces before and after the specified string. However, since there are no whitespaces before, the function will simply trim the whitespace after. We can also use the rtrim function to trim only the whitespaces at the end and ltrim function to trim only the whitespaces at the beginning.

These are only a few string functions and there are many available out there. Please practice with as many functions as you can.

Task 5 is complete!

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Task 6: Window Functions

Let us look at window functions, which are different from the built-in functions. These are the functions, which are applied over a window i.e., a set of rows rather than each row. For example, we can find out the salary for each employee by department and rank them from high to low or vice versa.

Let us perform this task in IntelliJ IDEA. However, we may choose to perform it in the Spark shell as well.

Step 1: Download the employee.csv file from the URL below. This file contains four columns: userId, movieID, rating, and timestamp.

employee.csv - http://bit.ly/2Z3atOx

Please save this file in IdeaProjects/Spark/chapter_8 by creating a new folder.

Step 2: Open IDE, right-click the training package that you have created in the previous exercise and hover over New and then click on Scala Class. When prompted, enter window as the name, click on the dropdown menu for Kind, and select Object. Enter the import as shown below.

import org.apache.spark.sql.SparkSession

import org.apache.spark.sql.expressions.Window

import org.apache.spark.sql.functions._

Next, we need to write a case class so that we can specify the schema for our fields.

case class Employee(name: String, number: Int, dept: String, pay: Double, manager: String)

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Here, we have created a case class and named it Employee by specifying the fields and its types.

Step 3: Now, write the main function and create the SparkSession object as shown below.

def main(args: Array[String]): Unit = {

val spark = SparkSession

.builder()

.appName("Creating a Dataset")

.master("local[*]")

.getOrCreate()

Next, let us import the implicits and load the file as shown below.

import spark.implicits._

val employeeDS = spark.read

.format("csv")

.options(Map("header" -> "true", "inferSchema" -> "true"))

.load("chapter_8/employee.csv")

.as[Employee]

Our program should look something like the one shown in the screenshot.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 4: Now that we have loaded our file, let us first create a window. We shall create a window to partition by the department and order by pay in the descending order.

val window = Window.partitionBy($"dept").orderBy($"pay".desc)

We have called the partitionBy and orderBy method on the Window object to create a window. The partitionBy method creates partitions for each department within the window and orderBy method orders the rows by pay in the descending order.

Step 5: Let us first use the rank function to get the pay for each employee by department in the descending order.

val ranked = rank().over(window)

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

We have simply called the rank function over the window we had created in the previous step by using the over method. The rank function allocates the increasing integer values to rows based on the column we specified in the orderBy method.

Let us now call the show method on our dataset and add the rank column by using the select function.

employeeDS.select($"*", ranked.as("rank")).show()

We should have the following output when we run this program.

Step 6: Let us now find the third highest salary from each department.

val rankedDS =  employeeDS.select($"*", ranked.as("rank"))

val second = rankedDS.where("rank = 3")

Let us now call the show method on our dataset.

second.show()

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

We can observe that we have obtained the third highest salary of employees from each department. Since Admin Offices department had two employees with the same pay, they both are ranked as three.

Step 7: Let us now look at dense_rank. Both these functions are similar and provide the increasing integer value based on the ordering of rows. However, there is a difference when it comes to dealing with the duplicate values. If there are two duplicate values, the rank function allocates the same rank for both the rows, skips the next rank for next row, and allocates the next incremental rank for the next row instead. For example, if rank 1 is allocated to the top two rows due to the presence of duplicate values, rank 2 will be skipped and instead, rank 3 will be allocated to the next row.

However, the dense_rank function does not skip the ranks when it encounters duplicate values. In the example above, the dense_rank will allocate the same rank i.e., rank 1 for duplicate rows and then rank 2 for the next row. Let us look at this now.

val denseRanked = dense_rank().over(window)

Let us now call the show method on our dataset and add the rank column by using the select function.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

employeeDS.select($"*", denseRanked.as("dense_rank")).show()

The following output is shown when we execute the program.

As a challenge, please find out the second highest salary from each department by using the dense_rank function.

Step 7: Similarly, let us look at the row_number and percent_rank.

The row_number function, as the name suggests, provides a row number for each row starting from 1 irrespective of the duplicates. The numbers are neither skipped nor are repeated.

val rowNumber = row_number().over(window)

employeeDS.select($"*", rowNumber.as("rowNumber")).show()

The following output is shown when we run the program.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

The percent_rank calculates the percent rank of a given row based on the following formula:

(rank - 1) / (the number of rows in the window or partition - 1)

For example, if the rank of a row is 11 and there are 101 rows in the partition, the rank is evaluated as 11-1/101-1 or 10/100 or 0.1. The percent rank ranges from 0 to 1. The first row will always have a percent rank of 0.

val percentRank = percent_rank().over(window)

employeeDS.select($"*", percentRank.as("percentRank")).show()

The following output is shown when we execute the program.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Step 8: Let us finally look at the lead and lag functions. The lead and lag functions are utilized to determine the value of leading or lagging of next row, when compared to the current row.

val leads = lead($"pay", 1, 0).over(window)

employeeDS.select($"*", leads.as("lead")).show()

The lead function takes three arguments. The first argument is the column name, second is the offset value, which determines the total number of rows preceding/succeeding the current row and the third argument is the default to specify the default value if the offset is outside the scope of the window.

The following output is shown when we execute the program.

Creating Ds  >> DF for RDD  >>  Agg & Collec

Date/Time.

Math & String

Window Functions

Similarly, there is lag function, which calculates the lag.

val lags = lag($"pay", 1, 0).over(window)

employeeDS.select($"*", lags.as("lag")).show()

The following output is shown when we run the program.

Task 6 is complete!

SUMMARY

In this chapter, we have looked at datasets API, which is an advanced Spark API. We learned about the datasets API, its applications, and its advantages over RDDs and DataFrames.

In the labs, we have had our hands on creating a dataset, converting an RDD to dataset by using the toDF method. Subsequently, we used the built-in functions to process data.

REFERENCES

● https://spark.apache.org/

● https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/functions.html

● https://spark.apache.org/docs/latest/api/sql/index.html

CHAPTER 9:

USER-DEFINED FUNCTIONS

Theory

In the previous chapter, we learned about the built-in functions. Let us now look at the User-Defined Functions.

Why User-Defined Functions?

Spark SQL has tons of built-in functions available to process data. These functions include aggregations, date-time functions, Math functions, String functions, collection functions and window functions. These built-in functions are applied in most of the cases. However, there can be scenarios, which cannot be accommodated by the usage of built-in functions. For such scenarios, we can write our own functions, which are called the User-Defined Functions. We simply have to write the codes of these custom functions and register them with Spark SQL to be able to use them in our programs.

We can classify such custom written functions in two types:

● User-Defined Functions (UDFs)

● User-Defined Aggregate Functions (UDAFs)

The difference between a UDF and a UDAF is that the UDF performs the user-defined operation on one row at a time and returns the result for each row, while UDAF performs the user-defined operation on a group of rows on a column. A few examples of built-in aggregate functions are sum, count, average etc., and they are followed by groupBy function most of the time.

Steps to implement User-Defined Function

All the User-Defined Functions (UDF & UDAF) can be implemented by following these three steps:

1. Writing a User-Defined Function.

2. Registering the User-Defined Function in Spark Application.

3. Using the User-Defined Function in Spark SQL or in the DataFrame API.

These steps are similar to implement a user defined function in Apache Hive and Apache Pig.

UDAF Types

The User-Defined Aggregate Functions (UDAFs) can be further classified into two types:

● Typed UDAFs

● Untyped UDAFs

The Typed UDAF is applied to datasets where we get the structure (schema) of rows by using the case class, whereas an untyped UDAF is applied to the DataFrames.

It is always preferable to use the built-in functions over the UDFs due to their standardization, as they are an efficient way to perform standard operations. Nonetheless, UDFs should only be developed and used when there is no other choice.

Do not worry if this does not make any sense right now. We shall be looking at UDF and UDAF in our lab exercise to understand this in a better way. The UDFs in Spark SQL are easy to implement. We can also use the Hive UDFs within the Spark SQL.

Before we jump into our lab exercises, let us learn some Scala concepts, which are required while writing our custom functions.

Function currying in Scala

Function currying in Scala is used to create partially applied functions. The curried functions are declared with multiple parameter groups with each group enclosed in parenthesis.

For example,

Scala> def sum(a: Int) (b:Int): Int = {

> a + b

> }

There can also be multiple parameters in each parameter group. Subsequently, these curried functions are used to create partially applied functions.

Partially applied functions in Scala

The partially applied functions, as the name suggests, are applied partially by only passing few parameters, while holding back other parameters. Hence, the partially applied functions are used when we do not want to pass all the parameters at once, but instead we can pass some parameters first and then the other parameters at a later stage.

For example, we can partially apply the function that we created above by using currying.

Scala> val part = sum(54)_

This expression will return us a function object called part.

We can now pass the parameter, which we held back as shown here.

Scala> part(6)

Finally, the result here will be 60, which is the sum. We shall look at these concepts in our lab exercises to understand them in a better way.

That is all in theory for this chapter. Let us now proceed to our lab exercises.

AIM

The aim of the following lab exercises is to write and use the user-defined functions in Spark applications.

The labs for this chapter include the following exercises:

● Defining Currying functions

● Using partially applied functions

● Writing User-Defined Function (UDF)

● Writing Untyped UDAF

● Using Untyped UDAF

● Using Typed UDAF

We need the following packages to perform the lab exercises:

● Java Development Kit (JDK)

● Scala

● Spark

LAB EXERCISE 8: USER DEFINED FUNCTIONS

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

1. Defining Currying functions

2. Using partially applied functions

3. Writing User-Defined Function (UDF)

4. Writing Untyped UDAF

5. Using Untyped UDAF

6. Using Typed UDAF

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Task 1: Defining Currying Functions

Before we start writing custom functions, let us look at the currying functions. We shall be working in the Scala shell for the next two tasks.

Step 1: Open the terminal and fire up the Scala shell by simply typing in Scala in the terminal. You should see the Scala prompt as shown in the screenshot below.

$ scala

Step 2: Let us now define a simple currying function with two parameter groups as shown below to understand the concept of curried functions.

scala> def sum(x: Int) (y: Int): Int = {

| x + y

| }

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

We have defined a function called sum, which adds two numbers. Instead of passing the parameters as one group, we have curried the parameters in two different parameter groups. Primarily, this will help us with partially applied functions.

Step 3: Moreover, we can also define currying functions with multiple parameters inside each parameter group as shown below:

scala> def sumProd(a: Int, x: Int) (b: Int, y: Int): Int = {

| a * b + x * y

| }

Step 4: We can also define a currying function in such a way that we can transform a function that takes two or more parameters into a function, which takes only one parameter.

scala> def prod(a: Int) = (b: Int) => a * b

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

As we can observe from the screenshot above, we have declared a prod function which takes only one parameter a, and returns another function which in turn takes another parameter b, and returns the result.

Step 5: We can simply pass the arguments with each argument inside a parenthesis as shown below.

scala> prod(54)(22)

Task 1 is complete!

Task 2: Using partially applied functions

Let us now see how we can utilize the curried functions and apply them partially.

Step 1: We have created a sum function in Step 1 of the previous exercise. Let us use that function and partially apply the parameters for it.

scala> val sumObj = sum(6)_

This will return us a function object as depicted in the screenshot below.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

The _ is used as a placeholder for the parameter, we are holding back. It indicates to the compiler that we are partially applying a function.

Step 2: We can then use the function object later to pass the parameter that we held back as shown below.

scala> sumObj(5)

Step 3: Similarly, let us partially apply the sumProd function, which we had created in the Step 2 of the previous task.

scala> val sumProdObj = sumProd(5, 6)_

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

We can then pass the held back parameters later as shown below.

scala> sumProdObj(7, 8)

The result is as shown in the screenshot above.

Step 4: We have also created a prod function in the previous task. We can also apply partial functions to that function as well. Please complete this step by yourself as a lab challenge.

Task 2 is complete!

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Task 3: Writing User Defined Function

Now that we have learned about curried and partially applied functions, let us look at how to write a UDF. We will be writing a UDF to decrease the rating of each movie by 0.5. For this purpose, we will be calling the UDF by using both Spark SQL and DataFrame APIs.

Step 1: Download the ratings_head.csv file from the URL below. This file contains four columns: userId, movieID, rating, and timestamp.

ratings_head.csv - http://bit.ly/2X3r2wb

Please save this file in the IdeaProjects/Spark/chapter_9 folder.

Step 2: Open IntelliJ IDEA, create a new Scala object and name it decrRatingUDF. We shall be using the following imports in our application:

import org.apache.spark.sql.SparkSession

import org.apache.spark.sql.functions._

Next, let us define our UDF by using the val keyword instead of using the def keyword.

val decrUDF = udf((input: Double) => input - 0.5)

The syntax to define a function by using val is a bit different than what we have been using so far with the def function. Here we are assigning a function literal to an immutable variable. In addition, we have not specified the return type for the function as we can make use of Scala type inference to take care of the return type.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

We are then passing the function literal inside the UDF function. The UDF function takes a column as a parameter and returns a column. Since we are passing the entire column of our dataset Ratings as input to the decrUDF function in the application, we are using this UDF function.

There are not many differences between the val and def keyword to define functions. Please check the link in references section to learn more about the val vs def discussion.

The program at this point should look like the screenshot as shown below.

Step 3: Let us now declare a case class for the dataset we will load as shown below.

case class Ratings(userId: Int, movieID: Int, rating: Double, timeStamp: String)

Next, define the main function, create a Spark Session and load the file as dataset as shown below.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

def main(args: Array[String]): Unit = {

val spark = SparkSession

.builder()

.master("local[*]")

.appName("Ratings Decrement UDF")

.getOrCreate()

Let us ensure to import the implicits before we load the file as dataset.

import spark.implicits._

val ratings = spark

.read

.format("csv")

.options(Map("InferSchema" -> "true", "header" -> "true"))

.load("chapter_9/ratings_head.csv")

.as[Ratings]

The program should now look something like the screenshot below.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Step 4: Let us call the UDF decrUDF by using the DataFrame API by using the select method, as shown below.

val ratingDecDf = ratings.select($"*", decrUDF($"rating").as("ratingDec"))

Here, we are selecting all the columns from our dataset and then adding one more column called ratingDec, which is obtained by applying our UDF on rating column.

Let us finally call the show method to check the output.

ratingDecDf.show()

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

The output should be displayed as shown in the screenshot below, when we execute the program.

Please see that we do not need to register the UDF using the udf.register method when we have declared a function literal within the UDF. However, we must register our UDF when we have not used the function literal and defined the UDF by using the def keyword. We shall look at this in the next few steps.

Step 5: Let us now apply the UDF inside the Spark SQL query. We may either choose to create a new Scala object to apply the UDF inside Spark SQL query or within this section.

We have used the same Scala object to apply the UDF by using the Spark SQL.

Step 6: Let us now define a function to decrease the rating by 0.5 by using the def keyword. Please define this function outside the main function as shown in the screenshot below

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

def decrUDF2(input: Double): Double = {

input - 0.5

}

Step 7: Let us now register this new UDF by using the partially applied function as shown below. We have used the underscore placeholder to specify that this is a partially applied function, for which the parameter will be passed later in the program.

spark.udf.register("decrUDF2", decrUDF2 _)

Let us also create a temporary table by using the createOrReplaceTempView function as shown below. Then, we can run our SQL queries over this table.

ratings.createOrReplaceTempView("ratings")

val ratingDecDf = spark.sql("select *, decrUDF2(rating) as ratingDec from ratings")

The program should look something like the screenshot given below.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Step 8: Let us finally call the show method and check the output.

ratingDecDf.show()

The following output has been observed.

With this, we have written, registered and used a UDF.

Task 3 is complete!

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Task 4: Writing Untyped UDAF

In the previous task, we have implemented a UDF. In this task, let us learn to implement an Untyped UDAF, which is used with DataFrames. They are complex when compared to the UDFs. Let us write a UADF function to calculate the average, which is a built-in function available in Spark, but let us follow the methodology just to understand the process of writing a UADF.

While we write a UADF, we need to implement few methods similar to what we had done in Lab exercise 5, Task 5 by Implementing Custom Accumulators.

Step 1: We shall be using the same file ratings_head.csv, which we had used in the previous task. Let us use the average UDAF which we are about to implement to calculate the average rating per user.

ratings_head.csv - http://bit.ly/2X3r2wb

Please make sure this file is available in the IdeaProjects/Spark/chapter_9 folder.

Step 2: Create a new Scala object and name it averageUDAF. Next, we will need the following imports to implement a UADF. Please note that we have not used some of these imports so far. We shall learn about them as we continue with our program.

import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}

import org.apache.spark.sql.types._

import org.apache.spark.sql.{Row, SparkSession}

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Now that we have the required imports, we need to extend our object to inherit the UserDefinedAggregateFunction method as shown below.

object averageUDAF extends UserDefinedAggregateFunction {

We have used the import org.apache.spark.sql.expressions. UserDefinedAggregateFunction, so that we can inherit its methods in our program.

The program at this point, should now look like the screenshot below.

Please ignore the error you see for the object name. This will go away once we implement all the required methods for UDAF.

Step 3: The first method is to specify the inputSchema i.e., the data types of the input parameters we will be passing for this UDAF.

def inputSchema: StructType = StructType(Array(StructField("inputColumn", DoubleType)))

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

The inputSchema method returns a StructType. The inputColumn is of DoubleType as the ratings column is of type Double and is enclosed in an array of StructField, which is in turn enclosed in StructType. We are assigning schema to the input by using the StructType and StructField methods, as we are working on an Untyped UDAF.

Step 4: Next, we have to specify the bufferSchema method, which informs us how the data is being aggregated when the tasks are working inside the executors.

To understand this in a better way, consider that we have ratings column with four rows as shown below.

Ratings

2.5

5.0

3.5

4.5

When we evaluate the average, we first have to compute the sum of all the rows in a column and their count. Subsequently, the sum divided by count gives the average. Therefore, the task, which computes the average of rows of the column, has to store the intermediate sum and the count of records in a buffer.

The buffer is first initialized as (0, 0) by implying the sum and count of ratings, respectively. Therefore, the task reads the first row and updates the buffer values as (2.5, 1) by adding the initial values, where 2.5 is the sum and 1 is the count. Next, when the task reads second row of ratings column, the buffer updates the values as (7.5, 2). Then, the buffer is updated as (11.0, 3) for the third row and (15.5, 4) for fourth row. Therefore, we need to provide the data type for the count and sum of buffer. We do this by using the bufferSchema method as shown below.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

def bufferSchema = StructType(Array(StructField("sum", DoubleType),StructField("count", LongType)))

We are specifying the bufferSchema as we did with the inputSchema using StructType and StructField. The sum is of DoubleType and count is of LongType.

Step 5: Let us now specify the data type of return value using the dataType method. Since the average that returns is of type Double, we specify the DoubleType as shown below.

def dataType: DataType = DoubleType

Step 6: Next, we need to specify if the function always returns the same output on the identical input by using the deterministic method. This is true by default.

def deterministic: Boolean = true

The program at this point should look like the screenshot shown below.

Step 7: Now that we have set our input, buffer and output schema, we have to initialize our buffer (0, 0) by using the initialize method.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

def initialize(buffer: MutableAggregationBuffer): Unit = {

buffer(0) = 0.00

buffer(1) = 0L

The initialize method takes buffer of type MutableAggregationBuffer and returns nothing. As its name suggests, this buffer is mutable and is used for aggregation purposes. The two columns in the buffer sum and count are initialized to 0. The first buffer is of type Double and second is of type Long.

Step 8: At this point of time, we have specified the schema for input, buffer and output. We have also initialized the buffer with zero. We now have to write the necessary logic, so that tasks know how to update the buffer.

We do this using the update method as shown below.

def update(buffer: MutableAggregationBuffer, input: Row): Unit = {

if (!input.isNullAt(0)) {

buffer(0) = buffer.getDouble(0) + input.getDouble(0)

buffer(1) = buffer.getLong(1) + 1

}

}

The update method takes two parameters. One is the buffer, which we have initialized, and the actual input (which is ratings in our case). The first buffer simply adds the ratings and the second buffer increments the count by 1 until all the records are processed. This update method is applied to every task processing this job. After the update method completes processing, there will be a final output buffer for each task. All we have to do is to merge the output of each task, which we will learn in the next step.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Step 9: To merge the output of all the tasks, we use the merge method as shown below.

def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {

buffer1(0) = buffer1.getDouble(0) + buffer2.getDouble(0)

buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)

}

The merge method takes two buffers i.e., two outputs from each task. It simply adds two outputs and stores them back in buffer1. This is performed for all the buffers.

Step 10: At this point of time, we have one single output from merge method. It contains the sum of all records and their count. All we have to do is to write the necessary logic to find out the average. The average is nothing but the sum divided by total count, which has been implemented using the evaluate method.

def evaluate(buffer: Row): Double = buffer.getDouble(0) / buffer.getLong(1)

With this, we have completed writing our UDAF. The error below the object name should be gone by now.  It should look like the screenshot below.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Task 4 is complete!

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Task 5: Using Untyped UDAF

Now that we have successfully learned to implement an Untyped UDAF, let us use it to find the average rating per user.

Step 1: Let us create a new object avgRatingUDAF within the UDAF, which we had implemented in the previous task. Next, write the main function and create a Spark Session object as shown below.

object avgRatingUDAF {

def main(args: Array[String]) {

val sparkSession = SparkSession.builder

.master("local[*]")

.appName("Average Rating UDAF")

.getOrCreate()

The program should now look like the screenshot as shown below.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Step 2: Let us now register the UDAF with Spark.

sparkSession.udf.register("averageUDAF", averageUDAF)

Step 3: Let us now read the file and create a temporary view on the DataFrame to write SQL queries.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

val ratings = sparkSession.read

.format("csv")

.options(Map("InferSchema" -> "true", "header" -> "true"))

.load("chapter_9/ratings_head.csv")

ratings.createOrReplaceTempView("ratings")

Step 4: Let us finally use the UDAF in our SQL query as shown below.

val average = sparkSession.sql("SELECT userId,  averageUDAF(rating) AS avgRating FROM ratings GROUP BY userId")

Let us check the result utilizing the show method as shown below

average.show()

The following output is shown.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

We have successfully written a UDAF, registered and used it in the Spark application.

Task 5 is complete!

Task 6: Typed UDAF

Finally, let us write and use a Typed UDAF, which is used on datasets where we assign the schema by using the case class thereby making it type safe. Let us write a Typed UDAF, which calculates the average of ratings as we had done in few previous tasks. The methods, which we used in an Untyped UDAF, are different as compared to the methods in a Typed UDAF.

Step 1: We shall be using the same file ratings_head.csv, which we have used in the previous task. Let us use the average UDAF which we are about to implement to calculate the average of all the ratings.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

ratings_head.csv - http://bit.ly/2X3r2wb

Please make sure this file is available in the IdeaProjects/ Spark/chapter_9 folder.

Step 2: Create a new Scala object and name it averageTypedUDAF. Next, we will need the following imports to implement a UADF. We have not used some of these imports so far. We shall learn about them as we continue with our program.

import org.apache.spark.sql.expressions.Aggregator

import org.apache.spark.sql.{Encoder, Encoders, SparkSession}

Next, we have to declare the case classes to specify schema for both input and buffer. However, we need not use StructType here as we did with an Untyped UDAF. As we will be loading the input as dataset and not DataFrame, the schema is associated with the case class.

case class Ratings(userId: Int, movieID: Int, rating: Double, timeStamp: String)

case class Average(var sum: Double, var count: Long)

The first case class Ratings specifies the input schema, whereas the second case class Average specifies the buffer schema. Please note that we have used the var keyword to define the mutable fields in Average buffer as the buffer keeps on updating when task process each row, as explained in previous tasks.

Now that we have the required imports and case classes defined, we need to extend our object to inherit Aggregator abstract class as shown below.

object MyAverageAggregator extends Aggregator[Ratings, Average, Double] {

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

The Aggregator abstract class takes three parameters, which are the input, buffer and output type. The input is Ratings, the buffer is Average and the output type is Double.

The program should now look like the one as shown in the screenshot.

Please ignore the error you see for object name. This will go away once we implement all the required methods for the UDAF as in the previous tasks.

Step 3: Let us now implement the methods from the Aggregator abstract class. The first method is the zero method, which initializes the buffer to zero. This is similar to initialize method, which we had utilized in the previous task.

def zero: Average = Average(0, 0L)

Step 4: Next, we have to implement the reduce method, which is similar to the update method used in the previous task. The reduce method provides the necessary logic to specify how the tasks should process the rows and columns of dataset.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

def reduce(buffer: Average, rat: Ratings): Average = {

buffer.sum += rat.rating

buffer.count += 1

buffer

}

The reduce method takes the buffer which is of type Average and the input ratings. Then, the buffer is updated for each row in the rating column along with the count. Once all the records are processed, the final value of the buffer for each task is returned.

At this point, we have the buffer outputs of each task. We now have to merge them to get the final sum and count of all the records.

Step 5: We now have to implement the merge function, so that the buffer outputs from all the tasks are merged.

def merge(b1: Average, b2: Average): Average = {

b1.sum += b2.sum

b1.count += b2.count

b1

}

This function simply adds the sum and counts of all the buffers and returns back the buffer.

The program at this point should look like in the screenshot given below.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Step 6: Next, similar to the evaluate method in the previous task; we have to implement the finish method. The finish method contains the logic to compute the average i.e., dividing the sum with count.

def finish(reduction: Average): Double = reduction.sum / reduction.count

Step 7: We now have to implement the encoders for the buffer and output values by using the bufferEncoder and outputEncoder. These encoders are required for the serialization purposes to translate between the Scala and Spark types.

def bufferEncoder: Encoder[Average] = Encoders.product

def outputEncoder: Encoder[Double] = Encoders.scalaDouble

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

The error below the object name should be gone by now as we have implemented all the methods required in order to create a Typed UDAF.

The program should now look like the one shown in the screenshot.

Step 8: Let us now use the Typed UDAF. Create a new object within the program as shown in the screenshot and name it avgTypedUDAF. Define the main function and create the Spark Session.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

object avgTypedUDAF {

def main(args: Array[String]) {

val sparkSession = SparkSession.builder

.master("local[*]")

.appName("Average ratings Typed UDAF")

.getOrCreate()

Next, load the file.

val ds = sparkSession.read

.format("csv")

.options(Map("InferSchema" -> "true", "header" -> "true"))

.load("chapter_9/ratings_head.csv")

.as[Ratings]

The program should look like the screenshot below.

Currying Function  >> Partially App Fun  >>  Writing UDF

Writing UUDAF

Using UUDAF

Typed UDAF

Step 9: Let us now make use of UDAF which have written in this task. We have to call the toColumn method on our UDAF and give it a name by using the name method as shown below.

val averageRating = averageTypedUDAF.toColumn.name("averageRating")

Let us now use the select method for the UDAF as shown below.

val avg = ds.select(averageRating)

Finally, let us call the show method and run the program.

avg.show()

The output, which shows the calculation of the average of all the ratings, is as shown in the screenshot below.

This completes the implemention of the Typed UADF task.

Task 6 is complete!

SUMMARY

In this chapter, we have looked at User Defined Functions (UDFs), which is the custom functions Spark. Moreover, we learned about UDFs and UDAFs, and their applications. We have also acquired some essential Scala programming concepts, which are called function currying and partially applied functions.

In the labs, we have had our hands on Scala function currying and partially applied functions. Finally, we utilized UDFs and the two types of UDAFs to process data.

REFERENCES

● https://spark.apache.org/

● https://alvinalexander.com/scala/fp-book-diffs-val-def-scala-functions

CHAPTER 10:

FILE FORMATS

Theory

In this chapter, let us learn about the various file formats we can use with Apache Spark. There is a wide variety of file formats supported by Spark, and each of these formats has its own advantages and disadvantages based on our requirements. So far, we have been dealing with CSV files during the course of this book to load and store data. Let us now shift the gears and learn more about various file formats, and their applications.

We can read and write files of various formats to spark by using Spark APIs. They are the RDD APIs and DataSource (DataFrame or dataset) APIs. Let us look at the various file formats we can use with Spark and then in the lab exercises, let us see how we can read and write these file formats by using the Spark APIs.

DataSource API

Before we learn the various file formats processed by Spark, let us look at the DataSource read and write API structure in detail.

Reading Data

We can easily read data to Spark from file systems, such as HDFS, Amazon S3, local file system etc. by specifying the path to the file(s). All the methods to read various files are implemented within the DataFrameReader class. We can access this class by calling read on the sparkSession object. Next, we need to specify the file format by using format method if our input format is other than parquet, as it is the default format in Spark. Subsequently, we can optionally specify options by using the option method (or options method with multiple options) inside a Map object. We can also optionally specify the schema by using the schema method. Finally, by using the load method, we specify the input path for the file. The syntax to read data is as shown below.

sparkSession.read

.format(“_”)

.option(“key”, “value”)

.schema(“_”)

.load(“_”)

The _ in the syntax represents the appropriate values. For example, the format method takes the format of the file like csv, json etc.

Let us now look at various read modes, which we can use as options while reading data.

Read Modes

While we read data, there could be times where we encounter malformed or corrupt records. We can utilize the read modes within the option method to specify how we would like to deal with them. There are three types of read modes, which are described below:

permissive

This mode sets all fields to  nulls when it encounters a malformed record and all such records are placed in a new field in a column called columnNameOfCorruptRecord. This is the default mode.

dropMalformed

This mode simply drops or removes the entire row when it encounters the malformed records.

failFast

This mode fails the job and throws an exception as soon as it encounters the malformed records.

We can set this mode by using the option method as shown below.

sparkSession.read

.format(“csv”)

.option(“mode”, “dropMalformed”)

.option(“inferSchema”, “true”)

.option(“header”, “true”)

.schema(newSchema)

.load(“/usr/local/some.csv”)

Writing Data

Similar to reading data, we can also write data to a file system. All the methods to write various files are implemented within the DataFrameWriter class. We can access this class by calling write on the DataFrame we would like to save. Subsequently, we have to specify the format by using the format method. Next, as we specified options with reading data, we can also specify the options by using the option (or options) method. Finally, we use the save method to save the file by providing a path.

The syntax to read data is as shown below.

dataFrame.write

.format(“_”)

.option(“key”, “value”)

.save(“_”)

Let us now look at various save modes we can use as options while writing data.

Save Modes

Unlike Read modes, the save modes do not deal with malformed records, but instead they provide us with options on what to do if there is already data available in the path where we intend to save data. There are four save modes available to write the data, which have been elaborated as follows:

append

This mode simply appends the output file to the list of files that are already available in the  save location.

overwrite

When  overwrite mode is used, Spark overwrites any existing data available in the save location. Please take caution while using this mode; and only utilize it if the already-available data is not required.

errorIfExists

This mode throws an error and terminates the write job if there is any data present in the  save location. This is the default mode used by Spark when writing data to a file system.

ignore

This mode simply ignores any already-existing data in the  save location. The data will not be saved and nothing will be changed or removed in the existing data.

This is the structure associated with reading and writing data by using the DataSource API. Let us now look at various file formats.

Text Files

Text files can be read and saved by using the Spark’s DataSource API. There are two methods available in the DataSource API to read text files, which are called the text and textFile methods. Both these methods are used to read the text files but the only difference is that the text method returns a DataFrame with partition columns, if any, while textFile method returns a dataset, thereby ignoring the partition columns. An example of reading a text file is as shown below.

spark.read.text(“/usr/local/files/sample.txt”)

or,

spark.read.textFile(“/usr/local/files/sample.txt”)

We can also read a text file by specifying the format in the format method as shown below.

spark.read.format(“text”).load(“/usr/files/sample.txt”)

Moreover, we can also save the DataFrame as text by using the DataFrameWriter class. However, we must ensure that there is only one column of type String while saving, or else the write will fail. Each row in the DataFrame becomes a line in the output file. An example of writing a text file is as shown below.

df.write.text(“usr/local/output/out.txt”)

We can also write a text file by specifying the format in the format method as shown below.

df.write.format(“text”).save(“/usr/local/out.text”)

CSV Files

Similar to the text files, we can also save the CSV files by using the DataSource API. Each line in the CSV file represents a single record. A CSV file contains a large number of options for processing. With these options, we can efficiently handle many tricky scenarios, such as corrupt records etc.

CSV files support compression, and they are human readable and splittable. However, these files are not nestable, and hence, they cannot hold complex data structures. The following is an example to read a CSV file.

spark.read

.format(“csv”)

.option(“inferSchema” -> “true”)

.option(“header” -> “false”)

.load(“/usr/local/files/sample.csv”)

Writing to CSV files from a DataFrame is also straight forward as shown below:

df.write

.format(“csv”)

.mode(“errorIfExists”)

.save(“usr/local/output/out.csv”)

JSON

The JavaScript Object Notation (or JSON) format is also one of the most popular file formats around. It is compressible, splittable and human readable. Moreover, it is nested, and it supports complex data structures. With Spark, we can load a single line JSON and a multi-line JSON. All we need to do is to specify an option for multi-line JSON. However, it is recommended to use the single-line JSON whenever possible. This makes it easier for appends when we are writing it back to JSON format. Similar to CSV, there is a wide variety of options to read and write data.

The following is an example of loading a JSON:

spark.read

.format(“json”)

.option(“multiLine” -> “true”)

.schema(“userSchema”)

.load(“/usr/local/files/sample.json”)

Moreover, we can also write to a JSON as shown below:

df.write

.format(“json”)

.mode(“overWrite”)

.save(“usr/local/output/out.json”)

Parquet Files

Parquet is a widely used file format in Spark due to the optimizations it provides for analytical processing, and is considered as one of the most efficient file formats for Spark. Moreover, it is column-oriented data structure and is the default file format in Spark. When no format is specified, Spark automatically processes them as Parquet. It is compressible and splittable, but is not human readable. Furthermore, it supports complex data structures, and it can be easily processed when columns are of struct, map or array types. This is not possible with JSON and CSV files.

Parquet has only a couple of options while reading and writing the data. The following is an example to read a Parquet file.

spark.read

.format(“parquet”)

.load(“/usr/local/files/sample.parquet”)

We can also read without specifying the schema as shown below.

spark.read

.load(“/usr/local/files/sample.parquet”)

The following is an example to write data.

df.write

.format(“orc”)

.mode(“errorIfExists”)

.save(“usr/local/output/out.parquet”)

ORC Files

ORC is yet another columnar file format, which stands for Optimized Row Columnar (ORC). There are no options for ORC files because Spark processes ORC file format very efficiently. Both ORC and Parquet are similar, but Parquet is specifically optimized for Spark, whereas ORC is optimized for Hive.

Moreover, ORC supports complex data structures; it is splittable and can be compressed. However, it is not human readable. The following shows an example of reading an ORC file.

spark.read

.format(“orc”)

.load(“/usr/local/files/sample.orc”)

These are the major file formats used by Spark. Let us look at these in action during the lab exercises.

RDD API

Let us now look at reading data by using RDD API.

Text Files

We can easily read and write plain-text files by using Spark’s RDD API. The text files can be compressed, splittable and are human readable. However, text files are not very efficient when compared to other file formats. Please note that the splitting might not always be possible when certain compression codecs are utilized.

In the RDD API, we utilize the textFile method on SparkContext object to read the text file and saveAsTextFile method to save the output as text file. We can save the text files by using a compression codec by simply passing the name of the compression codec as a parameter to the saveAsTextFile method.

When a text file is read by using textFile method, each line becomes a record in that RDD. Moreover, we can also read multiple text files in a directory as a single RDD by using wholeTextFiles method. This creates a paired RDD with the names of individual files as keys and the entire content of those individual files as values. This is very useful in scenarios where there are many small files. We can read them all in one go by using the wholeTextFiles method.

The syntax to read a text file is given as follows:

sparkContext.textFile(“/path/to/text/file.txt”)

We can read multiple text files as shown below.

sparkContext.wholeTextFiles(“path/to/files”)

Similarly, we can save an RDD as a text file, as shown below.

rdd.saveAsTextFile(“/path/to/save”)

Sequence Files

Sequence file is the popular Hadoop file format, which contains keys and values in the binary form. We can read the Sequence files by using the sequenceFile method on the SparkContext object, and write them by using the saveAsSequenceFile method on the RDD.

The syntax to read the file is given as:

sparkContext.sequenceFile(“/path/to/seq/file”, classOf[keyDataType], classOf[valueDataType])

As we can observe in the syntax above, we also have to specify the data types of keys and values of the Sequence file by using classOf[] object.

The syntax to write the file is given as:

rdd.saveAsSequenceFile(“/path/to/seq/output”)

Please note that we cannot directly read or save a Sequence file by using the DataSource API. However, we can convert the RDD loaded from the Sequence file to a DataFrame by using the toDF method. To save a DataFrame to the Sequence File, we must create a paired RDD from a DataFrame and then use the saveAsSequenceFile method.

Hadoop Files

We can also read the output of the Hadoop MapReduce jobs to Spark by using the hadoopFile and newAPIHadoopFile methods. The hadoopFile method is the old Hadoop API, whereas the newAPIHadoopFile method is the new Hadoop API.

The syntax to read a Hadoop file using the old Hadoop API is as follows.

sparkContext.hadoopFile[keyDataType, valueDataType, inputFormatClass](“/path/to/file”)

As we can note from the syntax above, we have to specify the data types of both keys and values and the Hadoop input format class for the input file.

An example of this is as follows.

sparkContext.hadoopFile[Text, LongWritable, TextInputFormat](“/usr/local/files/out/part-00000”)

Please note that we should import the required classes for the data types and input formats as required.

The syntax to read a Hadoop file by using the new Hadoop API is as follows.

sparkContext.newAPIHadoopFile(“/path/to/file”, classOf[inputFormatClass], classOf[keyDataType], classOf[valueDataType], conf)

In the new Hadoop API, we provide the path to an input file or the comma separated list of files, Hadoop input format class, key and value data types and finally the conf object. We can utilize the conf object to specify properties, such as delimiters.

An example of this is as follows:

sparkContext.newAPIHadoopFile(“/usr/local/output/part-0000”. classOf[TextInputFormat], class[LongWritable], class[Text], conf)

Please note that we should import the required classes for the data types and input formats as required.

Do not worry if this does not make sense at this moment. We shall look at this in our lab exercises and everything should start to make sense.

AIM

The aim of the following lab exercises is to read and write various file formats in Spark applications.

The labs for this chapter include various exercises to work on the following file types:

● Text Files

● CSV Files

● JSON Files

● Parquet Files

● ORC Files

● Hadoop and Sequence Files

We need the following packages to perform the lab exercises:

● Java Development Kit (JDK)

● Scala

● Spark

LAB EXERCISE 9: USING FILE FORMATS

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

1. Text Files

2. CSV Files

3. JSON Files

4. Parquet Files

5. ORC Files

6. Hadoop and Sequence Files

Task 1: Text Files

Let us begin this lab exercise by reading and writing text files in Spark by using both DataSource and RDD APIs. For ease and simplicity, let us perform these tasks in the Spark Shell.

RDD API

Let us first read and write text files to Spark by using the RDD API.

Step 1: Download a text file from the URL below and save it in the path IdeaProjects/Spark/chapter_10/. Please create new directories as required. The IdeaProjects folder is present in our Home folder.

treasure_island.txt - http://bit.ly/2LBFLtt

Step 2: Open the terminal and fire up the Spark shell. Let us load the text file by using the following code:

scala> val textData = sc.textFile(“IdeaProjects/Spark/chapter_10/treasure_island.txt”)

This code reads the data and creates an RDD[String] as shown below. We can read data from any file system, such as HDFS, AWS, and Azure etc., by providing the complete path or fully qualified URL of that particular file system. Subsequently, we can perform all the RDD operations or convert them to a DataFrame or dataset as required.

Step 2: Let us now write this back to the file system as shown below.

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

scala> textData.saveAsTextFile(“IdeaProjects/Spark/chapter_10/output”)

Let us now check if the save was successful. For this purpose, open a new terminal and check the contents by using the command below.

$ cat IdeaProjects/Spark/chapter_10/output/part-00000

Step 3: Now, let us see how we can load multiple files in a directory by using the wholeTextFiles method. For this purpose, please download the files available in the URL below:

books - http://bit.ly/2kupo5v

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

The folder should contain six different files. Please save them all in the IdeaProjects/Spark/chapter_10/ folder. You should now have the six files in this path: IdeaProjects/Spark/chapter_10/books.

Step 4: Let us read these files by using the wholeTextFiles method. It reads all the files present in books folder. Please switch back to Spark-shell and read the files by using the code below.

scala> val textFiles = sc.wholeTextFiles(“IdeaProjects/Spark/chapter_10/books”)

This will return us a RDD[String, String], which is a paired RDD as shown below.

This paired RDD contains the name of the files as keys and the entire content of files as values.

Step 5: Let us simply print the file names by using the keys method on textFiles RDD as shown below.

scala> textFiles.keys.collect.foreach(println)

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Furthermore, we can also get the values by using the values method. We can also perform all the operations, which we can apply on Paired RDDs such as mapValues, reduceByKey, and sortByKey etc.

This Paired RDD can again be saved to file system by using the saveAsTextFile method as usual.

Let us now use text files with the DataSource API.

DataSource API

Let us now load and save text files by using the DataSource API.

Step 1: Download a text file from the URL below and save it in the path IdeaProjects/Spark/chapter_10/. Please create new directories as required. The IdeaProjects folder is present in our Home folder.

ratings.txt - http://bit.ly/2lJcCQF

Step 2: Let us now load this file to Spark by using the Spark Shell with the following code.

scala> val ratings = spark

.read

.text(“IdeaProjects/Spark/chapter_10/ratings.txt”)

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Let us now check if the read was successful by calling the show method on the ratings DataFrame.

scala> ratings.show()

We can also use the textFile method as shown below.

scala> val ratings = spark

.read

.textFile(“IdeaProjects/Spark/chapter_10/ratings.txt”)

Using textFile ignores the partition directory names.

Step 3: Let us write this back to the file system as shown below.

scala> ratings

.write

.text(“IdeaProjects/Spark/chapter_10/output1”)

Please ensure that you only have one string column while you save the text file successfully. Moreover, make sure that the output directory (in this case, output1) does not exist before you perform the write action.

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Step 4: Use the following command to check if the save was successful. We will have to initiate the new terminal to run this command, as this will not be executed in Spark shell.

$ cat IdeaProjects/Spark/chapter_10/output1/part*

We should see the file saved as shown below.

Task 1 is complete!

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Task 2: CSV Files

Let us now learn the reading and writing of the CSV files to Spark. In the previous chapters, we have been reading and writing the CSV files. However, let us also see some of the many options that can be used while reading and writing CSV files.

Step 1: Download the file ratings.csv from the URL below and save it to the IdeaProjects/Spark/chapter_10 folder.

ratings.csv - http://bit.ly/2L8IEBS

Each line of this file represents one rating of a movie by one user, and it has the following format: userId, movieId, rating, timestamp.

Step 2: Let us now read this file to Spark from Spark shell by using few options.

scala> val data = spark

.read

.format(“csv”)

.option(“InferSchema”, “true”)

.option(“header”, “false”)

.option(“nullValue”, “Null”)

.load(“IdeaProjects/Spark/chapter_10/ratings.csv”)

Here, we have used a new option called NullValue, which replaces all the null values with the provided string, which is Null in this case. The default is “”. Please check the references section for all the options, which can be utilized while reading or writing the CSV files. All the options can be utilized in this way or inside a map object.

Subsequently, we can then call the show method as shown in the screenshot below to check if it was successful.

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Step 3: We can also utilize the modes we have learned in our theory. Let us see a relevant example.

scala> val dataNew = spark

.read

.format(“csv”)

.options(Map(“InferSchema” -> “true”

, “header” -> “false”

, “nullValue” -> “Null”

, “mode” -> “FAILFAST”))

.load(“IdeaProjects/Spark/chapter_10/ratings.csv”)

scala> dataNew.show()

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Step 4: Let us now write this DataFrame back to the file system in CSV format.

scala> dataNew.write

.format(“csv”)

.option(“sep”, “|”)

.save(“IdeaProjects/Spark/chapter_10/output2”)

Here, we have utilized an option called sep, which replaces the delimiter from comma to a pipe.

Step 5: Let us check if the save was successful, as we had desired.

$ cat IdeaProjects/Spark/chapter_10/output2/part*

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Task 2 is complete!

Task 3: JSON Files

Similar to the previous tasks, let us read and write JSON files. We shall be reading two types of JSON files: One is a single-line JSON and other is the multi-line JSON file.

Step 1: Download the file example_1.json from the URL below and save it to the IdeaProjects/Spark/chapter_10 folder.

example_1.json - http://bit.ly/2lRFI06

Step 2: The following code can be used to read the single-line JSON file.

scala> val jsonData = spark.read

.format(“json”)

.option(“multiLine”, “false”)

.load(“IdeaProjects/Spark/chapter_10/example_1.json”)

Step 3: Let us check if we were able to load the JSON file successfully.

scala> jsonData.show()

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Step 4: Let us now load the multi-line JSON file. Download the file example_2.json from the URL below and save it to the IdeaProjects/Spark/chapter_10 folder.

example_2.json - http://bit.ly/2lL3IST

Step 5: The following code is utilized to read the single-line JSON file.

scala> val multiJson = spark.read

.format(“json”)

.option(“multiLine”, “true”)

.load(“IdeaProjects/Spark/chapter_10/example_2.json”)

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Step 6: Let us now write this DataFrame to the file system.

scala> multiJson.write

.format(“json”)

.save(“IdeaProjects/Spark/chapter_10/output3”)

We can check the output by running the following command from a new terminal.

$ cat IdeaProjects/Spark/chapter_10/output3/part*

Task 3 is complete!

Task 4: Parquet Files

Parquet is Spark’s default file format. Let us learn to read and write Parquet files in Spark.

Step 1: Download the file userdata1.parquet from the URL below and save it to the IdeaProjects/Spark/chapter_10 folder.

userdata1.parquet - http://bit.ly/2kfIhJ4

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Step 2: Let us now read this Parquet file to Spark by using the code below.

scala> val parquetData = spark

.read

.load(“IdeaProjects/Spark/chapter_10/userdata1.parquet”)

Please see that we do not need to mention the format here, as Parquet is the default file format in Spark. However, we may explicitly mention the format as we did in the previous tasks if we desire so.

We should see the following output when we call the show method on the DataFrame.

Step 3: Let us write this back to the file system in the Parquet format.

scala> parquetData

.write

.save(“IdeaProjects/Spark/chapter_10/output4”)

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

We can check if the save was successful by simply running the cat command from a new terminal as shown below. However, we will not be able to read the file correctly as it is not a human-readable file.

$ cat IdeaProjects/Spark/chapter_10/output4/part*

Step 4: We can also save a parquet file by using the compression as shown below.

scala> parquetData

.write

.option(“codec”, “gzip”)

.save(“IdeaProjects/Spark/chapter_10/output5”)

Task 4 is complete!

Task 5: ORC Files

Step 1: Download the file userdata1_orc from the URL below and save it to the IdeaProjects/Spark/chapter_10 folder.

userdata1.orc - http://bit.ly/2kfQi0J

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Step 2: Reading an ORC file is similar to what we have been doing so far throughout this exercise.

scala> val orcData = spark

.read

.format(“orc”)

.load(“IdeaProjects/Spark/chapter_10/userdata1_orc”)

We should get the following output when we call the show method on the DataFrame.

Step 3: We can now simply write to an ORC format like any other file formats we have learned so far.

scala> orcData

.write

.format(“orc”)

.save(“IdeaProjects/Spark/chapter_10/output5”)

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Similar to Parquet, ORC is also not human readable and we only see the gibberish data when we use the cat command as shown below.

$ cat IdeaProjects/Spark/chapter_10/output5/part*

Task 5 is complete!

Task 6: Hadoop and Sequence Files

Let us now work with Hadoop and Sequence Files. These files are popular file formats with the Hadoop MapReduce framework. These files contain key value pairs in the binary format. Let us first create and write a Sequence file and then read the same sequence file.

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Sequence Files

Step 1: Let us first create an RDD by using the parallelize method as shown below.

scala> val seqRDD = sc.parallelize(List((“Ernesto”, 2000), (“Learning”, 4500), (“Lee”, “8000”)))

This creates an RDD[(String, Int)] as shown below.

Step 2: Let us now write the RDD to Sequence file format by using the saveAsSequenceFile method as shown below.

scala> seqRDD.saveAsSequenceFile(“IdeaProjects/Spark/chapter_10/seqOut”)

We may execute a cat command from another terminal to check if the save was successful, but the file will not be human readable as shown in the screenshot below.

$ cat IdeaProjects/Spark/chapter_10/seqOut/part*

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

We know that the save was successful by looking at SEQ at the beginning of the file. Moreover, we can also witness that the key type is of Text and the value type is of IntWritable.

Step 3: Let us now read the Sequence file we have just saved. Reading the Sequence files is a bit different from what we have been doing so far. While reading the Sequence file, we also need to specify the key and value data types.

scala> val seqData = sc

.SequenceFile(“IdeaProjects/Spark/chapter_10/seqOut/part-00000”

,classOf[org.apache.hadoop.io.Text]

,classOf[org.apache.hadoop.io.IntWritable])

As it is the Hadoop file format, we need to specify the data types in Hadoop. We have specified the Text and IntWritable types as the types for keys and values, since our keys are of String and values are of Int.

Step 4: However, since these are Hadoop data types, we cannot access the keys directly. We need to convert these types to Java data types as shown below. The job can fail if we do not convert the data types and collect.

scala> val newRDD = seqData.map

{

case (x, y) => (x.toString, y.get())

}

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

As we can observe from the screenshot above, we now have the RDD[(String, Int)]. We can now simply perform all the operations we usually do on RDDs. We have to utilize the toString method when converting from the Hadoop’s Text type and the get method for other data types.

Step 5: Let us now collect the RDD and check out the results.

scala> newRDD.collect()

With this, we have successfully written and read the Sequence files.

Hadoop Files

Hadoop files are the output of Hadoop MapReduce jobs. We can read these files with Spark and do further processing by using Spark. We shall now look at the old Hadoop API to read the output from MapReduce jobs to Spark.

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

Step 1: Download the file part-r-00000 from the URL below and save it to the IdeaProjects/Spark/chapter_10 folder.

part-r-00000 - http://bit.ly/2lSqdFy

This file is the output of a word count MapReduce job. It contains words as keys and values as the count separated by tab.

Step 2: Before we read the file, we first need the following imports. We need to import the data types for both keys and values and the input format. The keys are of type Text, values are Text and the input format is KeyValueTextInputFormat.

scala> import org.apache.hadoop.io.Text

scala> import org.apache.hadoop.mapred.KeyValueTextInputFormat

Step 3: Let us now read the file by using the hadoopFile API as shown below. This is the old Hadoop API.

scala> val hadoopData = sc.hadoopFile[Text, IntWritable, KeyValueTextInputFormat](“/IdeaProjects/Spark/chapter_10/part-r-00000”)

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

We now have an RDD from Hadoop MapReduce output. However, in order to access the key value pairs, we have to convert them to the Java data types first, as we did with the Sequence files.

Step 4: Convert the data types from Hadoop types as shown below.

scala> val hadoopRDD = hadoopData.map

{

case (x, y) => (x.toString, y.toString)

}

Step 5: Finally, let us call the collect method and check the output from the RDD.

scala> hadoopRDD.collect()

Text Files  >> CSV Files  >>  JSON files

Parquet Files

ORC Files

Hadoop & Seq Files

As we can observe from the screenshot above, we were successfully able to read the key-value pairs from the MapReduce output.

Please try to read the data by using new Hadoop API as a lab challenge.

Task 6 is complete!

SUMMARY

In this chapter, we have looked at various file formats processed by Spark. We have covered both the RDD API as well as DataSource API to read and write files from and to Spark.

In the labs, we have had our hands on using the various file formats in both the RDD API and DataSource API to read and write files from and to Spark.

REFERENCES

● https://spark.apache.org/

● https://spark.apache.org/docs/latest/sql-data-sources.html

● https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html

● https://docs.databricks.com/spark/latest/data-sources/read-csv.html

● https://docs.databricks.com/spark/latest/data-sources/read-json.html

CHAPTER 11:

SPARK CONFIGURATIONS

& OPTIMIZATIONS

Theory

Throughout the course of this book, we have been looking at Spark features and developing various applications. In this chapter, we shall look at important Spark configurations and optimization techniques to process our data more efficiently. Primarily, Spark has been built with the purpose to allow the developers to promptly start their development process and run various applications by default, as soon as Spark is installed. However, there are few scenarios where we have to modify configurations, fine tune settings and apply optimization techniques to run Spark applications faster with more efficiency.

Let us first look into Spark configurations followed by the performance optimization techniques.

Spark Configurations

Spark can be configured by using the following three options:

1. Spark configuration properties

2. Environmental variables

3. Logging

Let us now look at these options in detail and understand how they help us in configuring Spark.

Spark Configuration Properties

Almost every Spark configuration can be initiated by using the SparkConf class. Whenever a new SparkContext object is created, a SparkConf object is required. These configurations can be set in the following three ways:

● Configure Programmatically

● Set dynamically during runtime

● Configure the default spark-default.conf file

So far, we have been working with configurations programmatically to set application name and master etc. However, it is possible to set during runtime via command line, or set it in the spark-default.conf file. To specify the configurations during runtime, we have to make a jar file of our application and refer to it by using the spark-sumbit tool. Moreover, we can also use a Spark configuration file within the spark-sumbit tool. We shall look at this in the lab exercises and in our next chapter while we learn the running of Spark in a cluster.

With all these methods to set configurations, there is a precedence structure defined in Spark that determines which method has priority. The top priority is for the configurations set programmatically. Second priority is for the configurations defined by using the spark-submit tool. The third priority is for the configurations in the properties file, whereas the last priority is for the default values.

There are numerous types of configurations, which can be set within a Spark application. Let us look at few of the frequently used configurations in Spark and understand them in a better way. For all the Spark configurations, please check the link in the references section. Here are some common configurations in Spark.

Configuration Name

Default

Description

spark.app.name

None

This property is used to specify the name of a Spark application, which is reflected in the  Spark UI as well as logs.

spark.driver.memory

1g

This property is used to specify the amount of available memory for the driver process. The default value is 1 GB. The command line syntax for this property is --driver-memory.

spark.executor.memory

1g

This property is used to specify the amount of available memory for the driver process. The command line syntax for this property is given as, --executor-memory.

spark.driver.cores

1

This property is used to specify the number of cores for the driver process. The command line syntax for this property is --driver-cores.

spark.executor.cores

1

This property is used to specify the number of cores for each executor, and its command-line syntax is given as --executor-cores.

spark.master

None

This property is used to specify the cluster manager for  Spark.

spark.serializer

org

.apache

.spark

.serializer

.JavaSerializer

This property specifies the class used for serializing objects.

spark.eventLog.enabled

false

This property when set to  true enables the event log, which allows viewing completed Spark jobs in the history server.

Environment Variables

The environment variables for spark can be set through the conf/spark-env.sh file. The environment variables in Spark helps us set certain settings, such as java location. This spark-env.sh file is not available immediately after installing Spark but it can be copied from spark-env.sh.template file.

Please check the link in references section to find all the available environmental variables, which can be modified, within the spark-env.sh file. As a developer, we rarely change these settings. Hence, this section is included just for information purposes.

Logging

Log4j is used for logging in Spark. The entire configuration for logging process can be performed by using the conf/log4j.properties file. Similar to the environment variable file, log4j.properties file is also not available out of the box but can be copied by using the log4j.properties.template.

Performance Optimization

As Spark developers, we must always strive to run Spark applications as fast as possible with maximum efficiency. Although Spark is built with speed and efficiency as top priority, there are certain scenarios where we must tune the default settings and use some of the best practices to optimize our applications. Let us now look at few of these performance optimization techniques. These are only a few techniques to optimize our Spark application. These might not always work in every scenario. Keep experimenting by fine-tuning the settings until you find the best configuration that works.

Using Datasets extensively

Throughout this book, we have learned that the Dataset APIs provide the most efficient, faster and better error-handling measures. The Datasets APIs provide the best qualities of both RDDs and DataFrame APIs, and it is recommended to use the Datasets APIs wherever it is possible. Datasets (and DataFrames) are efficient and provide increased performance by using the Tungsten and Catalyst optimizations.

However, there could be scenarios where we have to utilize the RDD APIs to achieve something, which may not be possible via higher level APIs. In such cases, it is recommended to develop code in Scala or Java, as the serialization in these languages is much more optimized.

Avoiding UDF and UDAF

In order to better utilize these optimizations, it is recommended to use UDFs and UDAFs only when they are absolutely necessary. In addition, we should not develop UDFs and UDAFs, which perform more than one transformation or an action in a single UDF and UDAF. The goal should be to use built-in functions as much as possible, so that we can utilize the power of Spark to the fullest.

Furthermore, it is also recommended to develop UDFs and UDAFs in Scala or Java.

Data Serialization

Data serialization plays an important role in the performance of a Spark application. While the default Java serialization provided by Spark is efficient for almost all the cases, there are few cases where it is required to define a serialization for custom data types. In those cases where custom data types are used, we have to define them by using the Kryo serializer. Let us learn two Serializers provided by Spark in detail.

Java Serializer

Java  Serializers is the default Serializer in Spark. The objects in Spark are serialized by using the ObjectOutputStream framework. The Java Serialization works with any class that implements java.io.Serializable. It is also possible to tune the performance by extending the java.io.Externalizable class. Although this Serializer work with almost all the data types, it is quite slow and results in large serialized formats in many classes.

Kryo Serializer

Kryo Serializer is much faster and compact as compared to the Java Serializer. It can be used with Spark to serialize the custom data types. However, it does not support all the Serializable types; and it cannot be used without the prior registration of classes in an application.

Spark Memory Tuning

The memory of executors is allocated into the following two categories:

● Storage Memory

● Execution Memory

Storage Memory

The storage memory has a default allocation of 60% (0.6) of the total memory. The remaining 40% of space is reserved for user data, internal  Spark metadata and prevention of Out Of Memory (OOM) errors that may arise from very large datasets. This memory is used to cache or persist the datasets in memory when we call the cache() or the persist() methods. The amount of memory allocated for caching can be altered by using the spark.memory.fraction setting. The storage may occupy all the memory when there is no execution memory being used and vice versa. The execution memory may also evict the storage memory but only to a certain extent as described below.

Execution Memory

The execution memory is utilized to store the intermediate buffers while performing shuffles, joins, sorts and aggregation operations on the data. The amount of memory, execution may evict from the Storage memory allocation can be altered by using the  spark.memory.StorageFraction setting. The default is 0.5 of the total allocated Storage memory. This setting ensures that the amount of blocks, which have been reserved for storage, are never evicted.

This allocation of storage and execution memories ensures that when there is no caching required by the applications. Moreover, the entire memory can be used by execution and if the applications require caching, the required amount of space can be reserved, so that they are not evicted by the execution process. This dynamic allocation of memory helps utilize the memory efficiently in turn by providing faster results. This also ensures that the users need not worry about altering the memory allocations every time when they require more storage or caching memory. These default settings work most of the time but there might be scenarios where we might want to change these values as per requirement.

The following methods help tune the memory usage:

● Check the memory consumption of dataset by creating an RDD of the dataset and then look at the Storage page in the webUI. This will show the total consumption of memory by that RDD.

● The size of an object can be found by using the estimate method available in the SizeEstimator class. This is useful for determining the amount of heap space occupied by a broadcast variable each executor or the amount of space taken by each object when caching the objects in deserialized form. This is not the same as the serialized size of the object, which will typically be much smaller.

● Use Serialization while caching the dataset in memory. Please check the Caching topic in Chapter 5 for more information.

● Measure the impact of Garbage Collection by checking how often it runs, and how much time it takes. This can be achieved by adding -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps to the Java options property in Spark Configuration.

For example, spark.executor.extraJavaOptions=-XX:+PrintGCDetails -XX:+PrintGCTimeStamps

Once this property is set and submit a Spark job, we can see the messages printed in worker’s log each time a garbage collection occurs.  Please check the Spark documentation in References for more information on garbage collection.

Level of Parallelism

Parallelism denotes the number of tasks, which run in parallel on a single CPU core. It is recommended to start with three tasks per CPU core and then optimize the number, if required. The default level of parallelism can be modified by using the property spark.default.parallelism.

With a certain type of aggregation operations, such as sortByKey, groupByKey, reduceByKey, and join, etc., we might be greeted by a OutOfMemoryError. This is not because of low memory but because the working set of one of our tasks, such as one of the reduce tasks in groupByKey, was too large. By increasing the level of parallelism, we can have smaller working set of each task and thereby overcoming the OutOfMemoryError.

Levels of Data Locality

Data locality is a concept which refers to the distance between the node containing the data and the node containing the code to compute that data. It is logical to consider that the jobs run faster, if the data and the computation happens on the same JVM or node. It is always faster to move code to data than to move data to the code, as size of code is much smaller as compared to data.

The following are the levels of data locality from the closest to the farthest:

● PROCESS_LOCAL: This is the closest possible data locality where the data is available within the same JVM/Process/Executor as of code.

● NODE_LOCAL: At this level, the data and code are present within the same node but on a different process (executor). The data has to be moved between processes at this level and therefore there is a little overhead.

● RACK_LOCAL: In this case, the data and code are present in same rack but on different nodes. The data has to be moved through network at this level.

● ANY: The data at this level is not the same rack as of the code and could be elsewhere on a different rack/cluster/data center.

● NO_PREF: This level has no preference to data locality and the data is accessed from anywhere.

Spark thrives to launch the tasks at the best data locality level with the following priority: local -> node -> rack -> any. However, it is not always possible as the CPU might be busy processing another task and it has to wait sometime before it can acquire data at any other level before giving up. We can control the wait time for data locality by using the property that starts with spark.locality.wait and setting the value in time units. We can specify the wait time at each level by using the following properties:

spark.locality.wait.process - This property is used to set the wait time for the process locality. The default value of wait time is 3 seconds.

spark.locality.wait.node – This property is utilized to set the wait time for the node locality. The default value is 3 seconds.

spark.locality.wait.rack - This property is utilized to set the wait time for the rack locality. The default value is 3 seconds.

We can simply set ‘0’ to skip waiting and immediately move to the next level for any of these properties above. Please adjust the time for these settings as per the requirement to increase the speed and efficiency of Spark application.

Use Broadcast Variables

In Chapter 6, we have covered Broadcast variables, which are the best way to increase the performance of a Spark job. They are the shared variables, which allow Spark to send large values efficiently in an immutable (read-only) state to all the worker nodes. They can be used one or more times during the Spark operations. Moreover, they are sent to the worker nodes only once and are subsequently cached to the memory of the worker nodes in the deserialized form. These variables are very useful when the Spark job consists of multiple stages and multiple tasks in those stages require the same variable. Broadcast Variables overcome the inefficiency of shipping the variables every time to executors.

It is recommended to use the broadcast variables whenever possible to increase the overall performance of a Spark application.

Filter Data as soon as possible

To increase performance, filter your data as soon as possible. Always make sure that you have a filter operation before an operation, which would cause shuffling data over the network. For example, if you prefer to perform a join of two datasets and then filter the data, the shuffling of unnecessary data across the network is huge. However, if you filter the data first and then perform the join, lesser data is shuffled across the network.

Spark has in-built performance optimization for such cases called predicate pushdown. Spark automatically pushes the filter conditions as first operation in cases where the shuffling of data across network is involved.

Logs

Logs are the best way to get information on the performance of a Spark job. They are generated by driver and executor processes providing more insights on the job run. Logs have the information of errors and exceptions occurred during the execution of a job, and they are useful in troubleshooting and performance optimization techniques.

The amount of information the logs generate can be controlled by using the log4j properties file.

More Power

After trying most of these performance techniques, we may still observe low performance. Therefore, we can increase more computing power by adding additional nodes to the cluster.

The above-mentioned performance optimization techniques can help improve the performance of Spark applications if they are used wisely. However, these techniques might not always work in every scenario. Please use your best judgment on applying these techniques.

That is all in theory for this chapter. Let us now proceed to our lab exercises.

AIM

The aim of the following lab exercises is to configure Spark and learn various performance optimization techniques mentioned in the theory part of this chapter.

The labs for this chapter include the following exercises:

● Spark Configuration File

● Using spark-submit tool

● Environment Variables File

● Logging Properties File

● Checking Log Files

We need the following packages to perform the lab exercises:

● Java Development Kit (JDK)

● Scala

● Spark

LAB EXERCISE 10: SPARK CONFIGURATIONS & OPTIMIZATIONS

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

1. Spark Configuration Files

2. Using spark-submit tool

3. Environment Variables File

4. Logging Properties File

5. Checking Log Files

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Task 1: Spark Configuration File

Let us start this chapter by locating the Spark configuration file and modifying few of its properties.

Spark Configuration file contains all the default Spark configurations, which can be overridden by passing configurations at runtime or programmatically.

Step 1: The Spark configuration file is found in the Spark installation directory inside the conf folder. Open the terminal in your machine and enter the following command to change the working directory to the Spark conf directory in the terminal.

$ cd $SPARK_HOME/conf

You should see the full path of the directory as shown below. However, if you have not followed the lab exercises to install Spark from this book, you might have named the SPARK_HOME as something else. Please check Lab exercise 1 Step 4 for more information.

Step 2: Let us now check the contents of the conf directory by using the following command.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

$ ls

We should see the following files.

As we can see from the screenshot above, we have a file called spark-defaults.conf.template.

To be able to utilize the Spark configurations, we should either create a new file and name it spark-defaults.conf, or rename the spark-defaults.conf.template to spark-defaults.conf.

Step 3: Run the following command from terminal to rename the spark-defaults.conf.template to spark-defaults.conf.

$cp spark-defaults.conf.template spark-defaults.conf

Next, run the ls command again and you should see the file spark-defaults.conf, which we have renamed as shown in the screenshot below.

$ls

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Step 4: Let us now check the contents of Spark configuration file by running the following command. Please note that we can use any text editor to view this configuration file. We are using the gedit text editor as shown below:

$gedit spark-defaults.conf

We should now see the configuration file open in the text editor as shown below.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Scroll down a bit and you should see some example properties commented out with # sign at the start of the line as shown below:

Step 5: Let us now add some properties by removing # at the start of each property and add new ones as well.  Remove the # sign for the first property which is spark.master. Then change its value from spark://master:7077 to local as shown in the screenshot below.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

We have now set the default master for Spark as local in the Spark configuration file.

Next, let us add a new property to specify the Spark executor memory. To do this, go to the end of the configuration file and enter the following property and its value.

spark.executor.memory 4g

We have specified the memory for executors in Spark as 4 GB as shown in the screenshot below.

You can set the default configuration values for Spark by simply specifying the name of the property and its value separated by a white space.

Step 6: Once you have specified the required default configurations, you can save the file and close it. These values will be effective for the next Spark job you run. Please remember that these values can be overridden either dynamically or programmatically as explained in theory for this chapter.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Step 7: If you have followed the steps in this task and modified the configuration values, please delete the configurations file by using the command below. Please make sure that you delete the .conf file only and not the .template file.

$ rm spark-defaults.conf

Task 1 is complete!

Task 2: Using spark-submit Tool

Until now, we have been submitting jobs to Spark from the IDE by simply clicking the run button. This works well in the development phase, but when it comes to the later stages of software development cycles (such as production), it is not possible to trigger Spark jobs by using an IDE. We have to create a jar file of our application, move it to a node (usually the edge node) of the Spark cluster and then trigger the job by using the spark-submit tool via CLI.

Let us now look at how to trigger a job by using the spark-submit tool. We can also specify configuration properties here dynamically. Let us use one of the Spark programs we had written in the previous lab exercises. We shall be using the avgRatings.scala program for this task.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Step 1: First, we need to ensure that there are no hard-coded storage paths in our Spark program. This way we can supply the paths as arguments from the spark-submit tool. These paths are highlighted as shown below.

We can specify them as arguments which can then be provided at runtime by using the spark-submit tool. We shall look at this in detail in the next chapter. For now, we shall be using the program as is without changing any of the paths since we are running Spark application in the local mode only.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Step 2: Next, we need to export the jar file, so that we can run it by using the spark-submit tool. We shall be using the Scala Build Tool (SBT) to create a jar file, which contains our programs, and all the dependencies needed as one file. From the IDE, click on Terminal as shown in the screenshot below.

We should now see a command prompt as we would have in a traditional terminal at OS level.

Step 3: Navigate to the project’s folder with all the programs present by using the command below.

$ cd IdeaProjects/Spark/src/main/scala/training

Now, either create a build.sbt file here in this location by adding all the necessary dependencies, as we did while installing Spark or simply copy the build.sbt file from IdeaProjects/Spark/ folder.

We are simply copying the file here for simplicity. However, you are free to create a new build file.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

$ cp /home/{username}/IdeaProjects/Spark/build.sbt .

Please replace {username} with your username.

Step 4: Let us now use sbt to compile and then package a jar. To compile, run the following command.

$ sbt compile

Wait for few minutes for the compilation to finish. Afterwards, you should see the success as shown below.

Now run the following command to generate a jar file.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

$ sbt package

It will take some time to process and generate the jar file as required. Once it is finished, we should see a success message as shown in the screenshot below.

Step 5: Let us now verify if the jar has been successfully generated. We should now have two folders target and project. Inside the target folder, we will have a scala folder with its version and inside the scala folder, there will be a jar file created as shown below.

$ ls target/scala*

Now that we have successfully generated the jar file, we can copy it to any location or edge node (on Spark cluster) and run it via CLI by using the spark-submit tool.

For ease and simplicity, let us copy the jar file and the required file path which is hard coded in the program i.e., chapter_5/ratings.csv to the home folder.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

$ cd

$ cp IdeaProjects/Spark/src/main/scala/training/target/scala*/spark* .

$ cp -r IdeaProjects/Spark/chapter_5 .

All the files we require for spark-submit are now available. We can now run the spark-submit tool.

Step 6: Open a terminal from your desktop. You may close the IDE at this point of time. Run the following command to run the spark-submit tool:

$ spark-submit --class training.avgRatings spark_2.12-0.1.jar

Wait for a while for the job to finish processing.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Once it is done, we should see the result on the console as shown below.

While running the spark-submit tool, it is mandatory to specify the class of the Spark application by using the --class switch and then specifying the path to the jar file. Moreover, we can also specify other properties with the spark-submit tool. For example, we can specify the master and other configuration properties as shown below.

$ spark-submit \

--class training.avgRatings \

--master spark://85.67.45.54:7077 \

--deploy-mode cluster \

--executor-memory 50g

--num-executors 85

/path/to/spark_2.12-0.1.jar

We shall be looking at this again in the next lab exercises.

Task 2 is complete!

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Task 3: Environment Variables File

The environment variables file in Spark is used to configure the environment settings. They are the variables set outside of the Spark programs at the OS level. Let us now look at the environment variables file in Spark.

Step 1: Open the terminal and navigate to Spark’s conf directory. You should see the files as shown in the screenshot.

Currently, we are interested in the file called spark-env.sh.template.

Step 2: To be able to utilize the Spark environment variables file, we should either create a new file and name it spark-env.sh or rename the spark-env.sh.template to spark-env.sh.

$ cp spark-env.sh.template spark-env.sh

Next, use any of the text editors to open the file.

$ gedit spark-env.sh

We should see the file open as shown in the screenshot below.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Step 3: Scroll down the file and you can see all the environment variables commented with #. We can uncomment by removing the # sign before the variables and set the values. The screenshot below shows a few of the options we can set as the environment variables.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Step 4: We may modify the settings by reading the description of each setting. After the modification, save the file and the settings will be ready for the next Spark job.

Please consider deleting the environment variables file after you have saved it, as some settings might affect the framework adversely.

Task 3 is complete!

Task 4: Logging Properties File

Step 1: The log properties file is also found in the same conf directory of Spark installation. The file is called log4j.properties.template. As we performed previously, we need to rename the file to remove .template at the end. Please follow the steps from the previous exercises to achieve the same.

Step 2: Once you have the log4j.properties file, please use any of the text editors to view and edit the file. You should see the logging options as shown in the screenshot below.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Step 3: Let us set the log properties to save the logs to the disk. We can achieve this by adding the following properties to the log4j.properties file.

log4j.rootCategory=INFO,FILE

log4j.appender.FILE=org.apache.log4j.FileAppender

log4j.appender.FILE.File=IdeaProjects/Spark/logs log4j.appender.FILE.MaxFileSize=10MB

log4j.appender.FILE.MaxBackupIndex=10

log4j.appender.FILE.layout=org.apache.log4j.PatternLayout

log4j.appender.FILE.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L – %m%n

The log4j.properties file looks as shown below.

The above logging properties are utilized to enable the logs to a file by using the appropriate classes. We specify the path where the log files have to be saved along with its size and the layout.

Step 4: Save the file and these properties will be available when we run our next job.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

As opposed to what we did with the configuration files, please do not delete this file as we shall check the logs after running a Spark job in the next task of this exercise.

Task 4 is complete!

Task 5: Checking Log Files

For this task, let us run a Spark App and check if the log settings performed in the previous task work.

Step 1: Let us use the spark-submit tool to run a Spark application. This time we shall run the ratingsByMovies.scala program. First, let us copy the required input files to our home folder by using the following command:

$ cp -r IdeaProjects/Spark/chapter_6 .

Moreover, make sure to copy the jar we had generated in the previous task to the home directory if not already available.

Step 2: Let us now trigger the Spark application by using the spark-submit tool as shown below.

$ spark-submit --class training.ratingsByMovies spark_2.12-0.1.jar

The job should execute efficiently, and its output is given below:

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

Step 3: Now that we have successfully run the Spark job, let us check the logs file we had configured in the logs4j.properties file. In order to perform this process, change the current directory to Spark folder and open the logs file via any text editor.

$ cd gedit IdeaProjects/Spark/logs

We should see the logs persisted to a file as shown in the screenshot below.

Spark Conf  >> Spark Submit  >> Env. Variable

Log Prop.

Checking Log Files

This way we can check the logs to troubleshoot any errors or simply look for various options where we can enhance the overall performance.

Task 5 is complete!

SUMMARY

In this chapter, we have looked at various Spark Configurations. In addition, we have covered the Spark configurations, which include the Spark configuration properties, Environment Variables and Logging. Subsequently, we looked at various performance optimization techniques, which can help increase the efficiency and speed of a Spark application.

In the labs, we learned the procedures of setting up various configuration files and determined how to trigger a Spark job by using the spark-submit tool.

REFERENCES

● https://spark.apache.org/

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

● http://spark.apache.org/docs/latest/submitting-applications.html