Introduction to Apache Spark

​

Big Data Analytics with Python, AIMS2023

Dunstan Matekenya, PhD

Contents

1. What is Apache Spark?
2. Spark vs. Hadoop MapReduce?
3. Why Spark?
4. Getting and installing spark
5. Spark basic architecture
6. Sparks application concepts
7. Spark APIs and toolset
8. Hello Spark-First Spark program

1. What is Apache Spark?
2. Spark vs. Hadoop MapReduce?
3. Why Spark?

Whats Apache Spark

• We can just call it Spark for ease of use
• Spark is a unified engine designed for large-scale distributed data processing, on premises in data centers or in the cloud.
• Spark provides in-memory storage for intermediate computations, making it much faster than Hadoop MapReduce.
• It incorporates libraries with composable APIs for machine learning (MLlib), SQL for interactive queries (Spark SQL), stream processing (Structured Streaming) for interacting with real-time data, and graph processing (GraphX).

Spark’s Development History

• MapReduce major challenges:
• Not good for iterative computations such as machine learning
• Not good for interactive type of work such SQL-like queries or DataFrames which Data Scientists are accustomed to
• MapReduce style of processing data on disk rather than in memory makes it less conducive for other applications
• MapReduce programs are difficult to write
• In response to these challenges, several Apache open source projects were born: Apache Hive, Apache Storm, Apache Impala, Apache Drill, Apache Mahout and more
• Spark was one such project and it was started by researchers at UC Berkeley

Spark Vs. Hadoop MapReduce

​

Spark Vs. Hadoop MapReduce

• Overall, Spark is an enhancement of MapReduce
• Some of the enhancements include:
• make it highly fault tolerant and embarrassingly parallel
• support in-memory storage for intermediate results between iterative and interactive map and reduce computations
• offer easy and composable APIs in multiple languages as a programming model
• support other workloads in a unified manner.
• As opposed to the two-stage execution process in MapReduce, Spark creates a Directed Acyclic Graph (DAG) to schedule tasks and the orchestration of nodes across the Hadoop cluster.
• Spark focuses on its fast, parallel computation engine rather than on storage unlike Hadoop which combined the storage aspect HDFS as well as the compute engine

Spark Vs. Hadoop MapReduce Con’t

• Performance: Spark is faster because it uses random access memory (RAM) instead of reading and writing intermediate data to disks. Hadoop stores data on multiple sources and processes it in batches via MapReduce.
• Machine learning (ML): Spark is the superior platform in this category because it includes MLlib, which performs iterative in-memory ML computations. It also includes tools that perform regression, classification, persistence, pipeline construction, evaluation, etc.
• Scalability: When data volume rapidly grows, Hadoop quickly scales to accommodate the demand via Hadoop Distributed File System (HDFS). In turn, Spark relies on the fault tolerant HDFS for large volumes of data.

Source: IBM site

READ PLEASE

In Mining Massive Datasets, they provide a nice overview of differences between Spark and MapReduce. See page 41-48

Misconceptions About Spark and MapReduce

• Hadoop is cheap: Though it’s open source and easy to set up, keeping the server running can be costly. When using features like in-memory computing and network storage, big data management can cost a lot of money.
• Hadoop is a database: Though Hadoop is used to store, manage and analyze distributed data, there are no queries involved when pulling data. This makes Hadoop a data warehouse rather than a database.
• Hadoop is hard to set up: Though Hadoop management is difficult at the higher levels, there are many graphical user interfaces (GUIs) that simplify programming for MapReduce.

​

Source: IBM site

Why Choose Spark

• Fast-Sparks gets performance boost because of how it utilizes in-memory computation and use of directed acyclic graph (DAG) among other things
• User friendly-Spark has built structured data APIs such as DataFrames and SQL which has made accessing the functionality easy
• Modularity-Spark operations can be applied across many types of workloads and expressed in any of the supported programming languages: Scala, Java, Python, SQL, and R.
• Extensibility-Unlike Apache Hadoop, which included both storage and compute, Spark separates the two. You can use Spark to read data stored in myriad sources—Apache Hadoop, Apache Cassandra, Apache HBase, MongoDB, Apache Hive, RDBMSs, and more—and process it all in memory.

Getting, Installing and Running Spark

Spark Language APIs

• Scala: Spark is primarily written in Scala, making it Spark’s “default” language.
• Java: Even though Spark is written in Scala, Spark’s authors have been careful to ensure that you can write Spark code in Java.
• Python: Python supports nearly all constructs that Scala supports.
• SQL: Spark supports a subset of the ANSI SQL 2003 standard. This makes it easy for analysts and non-programmers to take advantage of the big data powers of Spark.
• R: Spark has two commonly used R libraries: one as a part of Spark core (SparkR) and another as an R community-driven package (sparklyr).

Spark’s language APIs make it possible for you to run Spark code using various programming languages.

Installing Spark for Single Use in Python

• If you are getting Spark just to use on your Laptop or Desktop, the best way is to install using instructions from the Spark website. As of Spark 3.0, you can install from Pypi using
• Spark is compiled for specific versions of Hadoop
• You need to make sure that you have Java JDK installed and JAVA_HOME variable set

Installing Spark for a Cluster

• Whether the cluster is on the premise or on the cloud. Installing Spark requires that you also install a cluster management system such as Hadoop or Kubernetes
• Since all the components of Hadoop and Spark are open source, you can do this on your own BUT its recommended to use distribution platforms
• Some of the popular distribution platforms include:
• AWS EMR
• Cloudera and Hortonworks
• MapR
• Databricks
• These platforms make installation easy and they also come with added functionality to make cluster management easy

Spark’s Basic Architecture

Overview of Spark Components

• We will refer to a Spark program as an Application
• A Spark application consists of a driver program that is responsible for orchestrating parallel operations on the Spark cluster.
• The driver accesses the distributed components in the cluster—the Spark executors and cluster manager—through a SparkSession.
• Therefore, the main components of a Spark application include:
• Spark Driver program
• SparkSession
• Spark executors
• Cluster manager

​

Apache Spark components and architecture

Spark Driver Program

• It is part of the Spark application which runs the driver process
• The driver process runs your main() function, sits on a node in the cluster
• Its first role is to instantiate a SparkSession
• It communicates with the cluster manager
• Requests resources (CPU, memory, etc.) from the cluster manager for Spark’s executors (JVMs)
• It transforms all the Spark operations into DAG computations, schedules them, and distributes their execution as tasks across the Spark executors.
• Once the resources are allocated, it communicates directly with the executors.
• Also responds to a user’s program or input

Another view of Spark’s Architecture

The SparkSession

• A SparkSession is one of the driver processes which controls the Spark Application and is the way Spark executes user-defined manipulations across the cluster.
• In Spark 2.0, the SparkSession became a unified conduit to all Spark operations and data.
• Through the SparkSession, you can create JVM runtime parameters, define DataFrames and Datasets, read from data sources, access catalog metadata, and issue Spark SQL queries.
• SparkSession provides a single unified entry point to all of Spark’s functionality.
• In a standalone Spark application, you can create a SparkSession using one of the high-level APIs.
• In the Spark shell (more on this in the next chapter) the SparkSession is created for you, and you can access it via a global variable called spark or sc.

​

The Cluster Manager

• The cluster manager controls physical machines and is responsible for managing and allocating resources for the cluster of nodes on which your Spark application runs.
• Currently, Spark supports four cluster managers:
• standalone-a simple cluster manager included with Spark that makes it easy to set up a cluster.,
• Apache Hadoop YARN-the resource manager in Hadoop 2.,
• Kubernetes-an open-source system for automating deployment, scaling, and management of containerized applications.

​

Spark Executor

• A Spark executor runs on each worker node in the cluster.
• The executors are responsible for actually executing the work that the driver assigns them. This means, each executor is responsible for only two things: executing code assigned to it by the driver and reporting the state of the computation, on that executor, back to the driver node.
• The executors communicate with the driver program and are responsible for executing tasks on the workers.
• In most deployments modes, only a single executor runs per node. In local mode (on a single computer, you can think of each core as an executor)

​

The Spark Web UI

• Apache Spark provides a suite of Web UI/User Interfaces (Jobs, Stages, Tasks, Storage, Environment, Executors, and SQL) to monitor the status of your Spark application, resource consumption of Spark cluster, and Spark configurations.
• Every time you initiate a Spark application, Spark will launch the UI on http://localhost:4040/
• Spark UI by default runs on port 4040 and below are some of the additional UI’s that would be helpful to track Spark application.
• Spark Application UI: http://localhost:4040/
• Resource Manager: http://localhost:9870
• Spark JobTracker: http://localhost:8088/
• Node Specific Info: http://localhost:8042/
• To access these URLs, Spark application should in running state.
• Refer to this link for details on Spark Web UI

Deploying Spark Applications

Because the cluster manager is agnostic to where it runs (as long as it can manage Spark’s executors and fulfill resource requests), Spark can be deployed in some of the most popular environments—such as Apache Hadoop YARN and Kubernetes.

​

Spark Deployment modes

Deployment mode distinguishes where the driver process runs. In "cluster" mode, the framework launches the driver inside of the cluster. In "client" mode, the submitter launches the driver outside of the cluster.

Other Spark Concepts Related to Spark Application

• Application jar: A jar containing the user's Spark application. In some cases users will want to create an "uber jar" containing their application along with its dependencies. The user's jar should never include Hadoop or Spark libraries, however, these will be added at runtime.
• Worker node: Any node that can run application code in the cluster
• Task-A unit of work that will be sent to one executor
• Job: A parallel computation consisting of multiple tasks that gets spawned in response to a Spark action (e.g. save, collect); you'll see this term used in the driver's logs.
• Stage: Each job gets divided into smaller sets of tasks called stages that depend on each other (similar to the map and reduce stages in MapReduce); you'll see this term used in the driver's logs.

Spark Language APIs-Revisited

• Spark’s language APIs allow you to run Spark code from other langauges.
• However, eacl language will maintain the same core architecture which we looked at
• There is a SparkSession available to the user, the SparkSession will be the entrance point to running Spark code.
• When using Spark from Python or R, the user never writes explicit JVM instructions, but instead writes Python and R code that Spark will translate into code that Spark can then run on the executor JVMs.

Spark Application Concepts

Distributed Data and Partitions

• Actual physical data is distributed across storage as partitions residing in either HDFS or cloud storage or other data stores
• A partition is a collection of rows that sit on one physical machine in our cluster. For instance, a DataFrame’s partitions represent how the data is physically distributed across your cluster of machines during execution.
• Partitioning allows for efficient parallelism. A distributed scheme of breaking up data into chunks or partitions allows Spark executors to process only data that is close to them, minimizing network bandwidth

Partitions

Each executor’s core gets a partition of data to work on

With DataFrames, we do not (for the most part) manipulate partitions manually(on an individual basis), however, we can still (and need to) control partitioning. For instance, specify number of partitions.

This code snippet will break up the physical data stored across clusters into eight partitions, and each executor will get one or more partitions to read into its memory

Summary of Key Concepts-Again

• Application-A user program built on Spark using its APIs. It consists of a driver program and executors on the cluster.
• SparkSession-An object that provides a point of entry to interact with underlying Spark functionality and allows programming Spark with its APIs. In an interactive Spark shell, the Spark driver instantiates a SparkSession for you, while in a Spark application, you create a SparkSession object yourself.
• Job-A parallel computation consisting of multiple tasks that gets spawned in response to a Spark action (e.g., save(), collect()).
• Stage-Each job gets divided into smaller sets of tasks called stages that depend on each other.
• Task-A single unit of work or execution that will be sent to a Spark executor.

Spark components communicate through the Spark driver in Spark’s distributed architecture

Spark Jobs

• The driver converts the Spark application into one or more Spark jobs
• It then transforms each job into a DAG.
• The DAG can be thought of as Spark’s execution plan, where each node within a DAG could be a single or multiple Spark stages.

Spark driver creating one or more Spark jobs

Spark Stages

• The driver converts the Spark application into one or more Spark jobs
• It then transforms each job into a DAG.
• The DAG can be thought of as Spark’s execution plan, where each node within a DAG could be a single or multiple Spark stages.

Spark driver creating one or more Spark jobs

Spark Stages

• As part of the DAG nodes, stages are created based on what operations can be performed serially or in parallel.
• Not all Spark operations can happen in a single stage, so they may be divided into multiple stages.

Spark job creating one or more stages

Spark Tasks

• Each stage is comprised of Spark tasks (a unit of execution), which are then federated across each Spark executor.
• Each task maps to a single core and works on a single partition of data As such, an executor with 16 cores can have 16 or more tasks working on 16 or more partitions in parallel, making the execution of Spark’s tasks exceedingly parallel!

Spark stage creating one or more tasks to be distributed to executors

Transformations and Actions

• In Spark, the core data structures are immutable meaning they cannot be changed once created.
• In order to “change” a DataFrame you will have to instruct Spark how you would like to modify the DataFrame you have into the one that you want. These instructions are called transformations
• Transformations, therefore transform a Spark DataFrame into a new DataFrame without altering the original data, giving it the property of immutability.
• For instance, an operation such as select() or filter() will not change the original DataFrame; instead, it will return the transformed results of the operation as a new DataFrame.

Spark operations on distributed data can be classified into two types: transformations and actions.

Transformations and Actions

​

• Transformations allow us to build up our logical transformation plan.
• To trigger the computation, we run an action. An action instructs Spark to compute a result from a series of transformations.
• The simplest action is count, which gives us the total number of records in the DataFrame:

Transformation vs actions in Spark

Narrow Vs. Wide Transformations

• Transformations consisting of narrow dependencies are those where each input partition will contribute to only one output partition.
• A wide dependency style transformation will have input partitions contributing to many output partitions. You will often hear this referred to as a shuffle where Spark will exchange partitions across the cluster.

Narrow versus wide transformations

Narrow Vs. Wide Transformations

​

• Any transformation where a single output partition can be computed from a single input partition is a narrow transformation. For example, in the previous code snippet, filter() and contains() represent narrow transformations because they can operate on a single partition and produce the resulting output partition without any exchange of data.
• However, groupBy() or orderBy() instruct Spark to perform wide transformations, where data from other partitions is read in, combined, and written to disk. Since each partition will have its own count of the word that contains the “Spark” word in its row of data, a count (groupBy()) will force a shuffle of data from each of the executor’s partitions across the cluster. In this transformation, orderBy() requires output from other partitions to compute the final aggregation.

​

Lazy Evaluation

• All transformations are evaluated lazily. That is, their results are not computed immediately, but they are recorded or remembered as a lineage.
• In other words, in Spark, instead of modifying the data immediately when you express some operation, you build up a plan of transformations that you would like to apply to your source data
• Lazy evaluation is Spark’s strategy for delaying execution until an action is invoked or data is “touched” (read from or written to disk).
• An action triggers the lazy evaluation of all the recorded transformations.
• While lazy evaluation allows Spark to optimize your queries by peeking into your chained transformations, lineage and data immutability provide fault tolerance.
• The actions and transformations contribute to a Spark query plan, which we will cover in the next chapter. Nothing in a query plan is executed until an action is invoked.

Lazy Evaluation Illustrated

Lazy transformations and eager actions

The Spark UI

• Spark includes a graphical user interface that you can use to inspect or monitor Spark applications in their various stages of decomposition—that is jobs, stages, and tasks.
• Depending on how Spark is deployed, the driver launches a web UI, running by default on port 4040, where you can view metrics and details such as:
• A list of scheduler stages and tasks
• A summary of RDD sizes and memory usage
• Information about the environment
• Information about the running executors
• All the Spark SQL queries
• In local mode, you can access this interface at http://<localhost>:4040 in a web browser.

​

​

A Tour of Spark APIs and Toolset

Spark APIs

• Within each programming language, Spark has APIs which essentially is different ways to interact with Spark functionality depending on your use case.
• Spark has two fundamental sets of APIs: the low-level “unstructured” APIs, and the higher-level structured APIs.
• On top of these primitives (building blocks) Spark has a series of standard libraries for additional functionality.

Apache Spark Ecosystem of APIs and Libraries

Apache Spark components and API stack

SparkSQL

• This module works well with structured data. You can read data stored in an RDBMS table or from file formats with structured data (CSV, text, JSON, Parquet, etc.) and then construct tables in Spark.
• When using Spark’s Structured APIs in Java, Python, Scala, or R, you can combine SQL-like queries to query the data just read into a Spark DataFrame. For instance, in code below, read data as JSO but interact with it using SQL

SparkMLlib

• Doing on ML on large scale data comes with its own problems and Spark solves this problem by having a dedidated MLlib
• Spark comes with a library containing common machine learning (ML) algorithms called MLlib
• MLlib provides many popular machine learning algorithms built atop high-level DataFrame-based APIs to build models
• This package allow you to extract or transform features, build pipelines (for training and evaluating), and persist models (for saving and reloading them) during deployment.
• Additional utilities include the use of common linear algebra operations and statistics. MLlib includes other low-level ML primitives, including a generic gradient descent optimization.

​

Spark Structured Streaming

• Due to common use cases of streaming data, Spark introduced the Streaming API
• This API can interact with streaming engines such as Apache Kafka and other streaming sources, the new model views a stream as a continually growing table, with new rows of data appended at the end
• Developers can merely treat this as a structured table and issue queries against it as they would a static table.
• The API supports different streaming data sources such as Kinesis, and HDFS-based or cloud storage in addition to Apache Kafka

GraphX

• GraphX is a library for manipulating graphs (e.g., social network graphs, routes and connection points, or network topology graphs) and performing graph-parallel computations
• It offers the standard graph algorithms for analysis, connections, and traversals, contributed by users in the community: the available algorithms include PageRank, Connected Components, and Triangle

Further Reading

1. Look at the Apache Spark website

​

First Spark Program

Spark in Local Mode

• For all the tutorials in this class, we will run Spark in local mode where all the processing is done on a single machine
• We can do this in in a Spark shell—this is an easy way to learn the framework, providing a quick feedback loop for iteratively performing Spark operations.
• We can also run in local mode in Jupyter Notebook or using a standalone Python scripts. In this case, even though its local mode, we can still assign multiple cores to Spark depending on out compute resources
• Obviously, Spark is meant to be run on compute clusters
• So, for large data sets or real work where you want to reap the benefits of distributed execution, local mode is not suitable—you’ll want to use the YARN or Kubernetes deployment modes instead on a cluster on the cloud or on-premise

​

Spark Through Pyspark Shell

1. Step-1: Download Spark binaries from this page
2. Download the default option shown in the screenshot below
3. Extract the zip files and examine the contents

​

Spark Folder Contents

• The most import directory to note is the bin.
• This directory, as the name suggests, contains most of the scripts you’ll employ to interact with Spark, including the Spark shells (spark-sql, pyspark, spark- shell, and sparkR). We will use these shells and executables in this directory later in this chapter to submit a standalone Spark application using spark- submit

Running Spark Shell

1. First, cd to the bin directory
2. Once there, run pyspark
3. You can also run shell using spark-shell which runs shell with scala but lets stick with the Python shell
4. Exercise: Run the following commands in Pyspark shell
1. strings = spark.read.text("../README.md")
2. Check the data type for variable strings
3. strings.show(10, truncate=False)
4. strings.count()
5. Use exit() or Ctrl-D (i.e. EOF) to exit Spark shell

We used the high-level Structured APIs to read a text file into a Spark DataFrame rather than an RDD.

From now on, we will run Spark in

1. Jupyter Notebook
2. If possible, we will try to run in a cloud platform
3. Run in a Hadoop cluster using our little Sandbox