PYSPARK

PRESENTED BY
Naresh Kumar
01

• Apache Spark – Different Types of Execution Modes

• Apache Spark – Local Execution Mode
• Types of Cluster Managers
• Apache Spark – Standalone Execution Mode
• Apache Spark - Cluster Key Components

AGENDA
• Spark Runtime Architecture - Cluster Mode
• Spark Runtime Architecture - Client Mode
• Spark Application - Cluster Mode Vs Client Mode

WEEK-2 • Spark Driver Memory Allocation

• Spark Executor Memory Allocation
• Spark Executor Memory Allocation Deep-Dive
• Scenario-1: What happens when a user submits the spark application
• Scenario-2: What happens when a user submits the spark application
• How to calculate number of CPU cores required to process data in spark
• How to calculate number of executors required to process data in spark
• How much each executor memory is required to process data in spark
• How to calculate the total memory required to process data in spark
Spark Execution Modes

Local Mode

Standalone Mode

Cluster Mode

Local Mode: Spark runs on a single machine or node, within a single JVM process, suitable for development and testing with small datasets.
Computation is not distributed, and everything is handled locally.

Standalone Mode: Spark uses a built-in cluster manager, with a master node managing resources and worker nodes executing tasks. It
provides a simple setup for distributed processing without relying on external resource managers.

Cluster Mode: Spark runs in a distributed environment with a dedicated master node managing resources and multiple worker nodes
executing tasks. It is used for production workloads and large-scale data processing.

02
 Spark Execution : Local Mode

Local Mode: Local mode is the default Spark mode and requires no resource management.
When you use the spark-shell command. It's ideal for testing, quick setups, and handles
partitions equal to the number of CPUs on the machine.

You can initiate local mode using commands such as:

By default, spark-shell runs in local mode with a single thread. You can specify the local[N] attribute to run Spark with N threads, leveraging parallel
computation. Spark is optimized for parallel processing, so adjusting the number of threads allows multi-threaded execution in local mode.

03
Types of Spark Cluster Managers

In Apache Spark the cluster manager is responsible for allocating resources and managing the cluster on which the Spark application runs.
Spark supports various cluster managers like Kubernetes, Apache Mesos, Hadoop YARN, and standalone cluster manager.

❑ Standalone cluster manager (Spark's own manager that is deployed on private cluster)

❑ Hadoop YARN (Yet Another Resource Negotiator) cluster manager

❑ Kubernetes cluster manager

❑ Apache Mesos cluster manager (Deprecated)

Responsibilities of a Cluster Manager

▪ Resource Allocation: Allocates CPU, memory, and other resources to Spark applications.
▪ Monitoring: Keeps track of resource usage, application status, and node health.
▪ Fault Tolerance: Handles node failures and ensures applications continue running.

04
 Spark Execution : Standalone Mode

Standalone Mode: It is a built-in cluster manager that allows Spark to run on a cluster of machines
without external resource managers. It includes a master node for resource management and worker
nodes for task execution. This mode is easy to set up and manage, providing a straightforward
solution for distributed processing.

To submit a Spark application to a Standalone cluster, you use the spark-submit command with the --
master option set to the Spark master URL. Here’s how you can do it:

spark-submit --master spark://<master-hostname>:7077 <your-application>.jar

Example: Assume your Spark master is running on [Link], and your application
JAR file is [Link]. The command to submit the application would be:

spark-submit --master spark://[Link] [Link]

The above command submits the Spark application to the Standalone cluster, where the master node manages resources and schedules tasks on the
worker nodes.

05
 Spark Cluster – Key Components

The architecture of an Apache Spark cluster consists of a Master node, Worker nodes, and Executors.

Master Node: The master node is the central coordinator of the Spark cluster. It manages the allocation of resources.

Worker Nodes: Worker nodes are the compute nodes in the Spark cluster where actual data processing tasks are executed.

Executors: Executors are JVM processes that run on worker nodes and execute tasks as directed by the driver node. They
are responsible for processing data, storing intermediate results, and returning final results to the driver.

Resource Manager (RM):

Manages the overall resources of the cluster.
Allocates resources to various applications.

Application Master (AM):

Manages the lifecycle of the application
Coordinates with the Node Managers (NMs) to launch containers (which can run executors).

KEY CONCEPTS:
✓ YARN is comprised of Resource Manager and Node Manager
✓ There is only one Resource Manager which runs on Master Node
✓ There will be multiple Node Managers running on each Data Node or Slave Node or Worker Node
✓ Resource Manager deals with resource management to execute any Job/Application
✓ Node Manager takes care of individual tasks/processes submitted to them 06
 Spark Runtime Architecture :: Cluster Mode

Edge/Client Node

M RM
RM W1 NM W2 NM

1
YARN E E
E E
Spark-submit
(Cluster Mode)

2 3
E Available Executor W3 NM W4 NM W5 NM

E Occupied Executor AM
Driver E E E E
SC

AM - Application Master SC - SparkContext

07
 Spark Application Submission :: Cluster Mode

User submits the spark application to cluster manager (YARN-Resource Manager).

Resource Manager(RM) starts the Application Master in any one of the worker node inside the cluster.

Application Master will launch the driver on the same worker node in cluster.

Driver creates a SparkContext (SC ) on the same worker node.

SC requests RM and acquires the required executors. And these executors will register with Driver.

Driver converts User Code -> Logical DAG -> Physical DAG-> Creates Tasks.

Spark Context sends all tasks to the executors.

Driver will keeps the track of all the executors running status and get the results from them. And on successful
completion of all the tasks, executors will unregister with the Driver.

Driver sends the results to the respective location as per the code and close the application.

08
 Spark Runtime Architecture :: Client Mode

M RM
RM W1 NM W2 NM
Edge/Client Node

YARN E E
Driver E E
SC

Spark-submit
(Client Mode) W3 NM W4 NM W5 NM
E Available Executor

E E E E E E
E Occupied Executor

AM - Application Master SC - SparkContext

09
Spark Application Submission - Client Mode - Points to remember

✓ when you submit an application in client mode, the application master is not created. Instead, the driver program, which acts as the application
master, runs on the client machine from which the Spark application is submitted.

✓ Driver program is responsible for coordinating the execution of tasks and interacting with the cluster manager.

✓ Driver program communicates directly with the cluster manager to request resources.

✓ Once resources are allocated, the driver launches executors on worker nodes to perform the actual data processing.

✓ Throughout the execution lifecycle of the Spark application, the driver program remains active on the client machine, coordinating task
execution, collecting results, and handling any failures or exceptions that may occur.

✓ In client mode, resources such as CPU, memory, and network bandwidth on the client machine are utilized by the driver program for managing
the Spark application. The client machine should have sufficient resources to handle the workload of the Spark application.

10
Spark Application - Cluster Mode Vs Client Mode

Client Mode Cluster Mode

Driver will be created in client node outside of the cluster Driver will be created in worker node inside the cluster

Network latency is high Network latency is low

Logs are generated in client machine. Hence, easy to debug. Logs are generated in std out/std err file. Hence, extra effort is required to
debug the errors.

Driver will die when client node get disconnected from the cluster. And then Driver will not die when client node get disconnected from the cluster. All
all the executors will also killed. the executors will run without issues.

High chances of Driver OOM Exceptions Low chances of Driver OOM Exceptions

11
Driver Memory Allocation

Master Node Worker Node-1 Worker Node-2

Spark-submit Driver Container
(Cluster Mode)
Executor Container

Yarn 2 CPU Cores 4 CPU Cores

4 GB RAM 8 GB RAM

JVM Process
★ Non JVM Overhead memory = 10% or 384MB(whichever is higher) of [Link] = 4GB JVM heap Memory
container memory.
★ Overhead memory will be consumed by container processes, shuffle
exchange and other its internal usage. Non JVM Process
[Link]=400MB
Overhead Memory
★ In sufficient Overhead or JVM heap memory can cause to spark Out Of
Memory(OOM) Exception.

12
Executor Memory Allocation

Master Node Worker Node-1 Worker Node-2

Spark-submit Driver Container
(Cluster Mode)
Executor Container

Yarn 2 CPU Cores 4 CPU Cores

4 GB RAM 8 GB RAM

JVM Process
★ Non JVM Overhead memory = 10% or 384MB(whichever is higher) of [Link] = 8GB JVM heap Memory
container memory.
★ Overhead memory will be consumed by container processes, shuffle
exchange and other its internal usage. Non JVM Process
[Link]=800MB
Overhead Memory
★ In sufficient Overhead or JVM heap memory can cause to spark Out Of
Memory(OOM) Exception.

13
 Executor Memory deep dive

[Link]=0.6 [Link]=0.5

300MB
300MB Reserved Memory Reserved Memory used for Spark's internal operations

used for caching and persisting data

Executor The storage memory can store RDDs,

(60% of 7.7GB)
Storage Memory Pool DataFrames, and Datasets,

4.62GB
It is also dynamically shared with
execution memory, allowing for flexible
Spark Memory allocation based on workload demands.
JVM heap Memory
Used for Dataframes compute operations like
Executor Memory Pool join, aggrations,shuffles, transformations,

It's dynamically shared between execution

and storage, meaning if execution memory is
Overhead Memory
not being used, it can be allocated to
storage, and vice versa

Used for
User Memory User Memory UDF,
[Link] = 8GB broadcast variables,
objects that are not managed by Spark's
internal operations,
RDD conversion operations
[Link]=800MB

14
Note: If the executor memory is less than 1.5 times of reserved memory, Spark will fail with a “please use larger heap size” error message.
Example-1: What happens when a user submits the spark application

spark-submit \
--class [Link] \ 4GB-300MB(Reserved memory) = 3.7GB per executor Worker Node
--master yarn \ Worker Node
--deploy-mode cluster \
--driver-memory 4g \
--num-executors 3 \
--executor-memory 4g \ Driver
--executor-cores 2 \
/path/to/your/spark-application_2.[Link]

Worker Node
Master Node
Worker Node
Executor-1 Executor-2

1.85GB 1.85GB 1.85GB 1.85GB

YARN 1C 1C 1C 1C
Spark-submit
(Cluster Mode)

Worker Node

Executor-3

1.85GB 1.85GB
1C 1C

Note:. By default, Spark creates one partition for each block of the file (blocks being 128MB by default in HDFS), but you can also ask for a higher number of partitions by passing a larger value
15
Example-2: What happens when a user submits the spark application

spark-submit \
--class [Link] \
--master yarn \ 2GB-300MB(Reserved memory) =1700 MB per executor
--deploy-mode cluster \
--driver-memory 2g \
--num-executors 2 \
--executor-memory 2g \
--executor-cores 2 \
/path/to/your/spark-application_2.[Link]
Worker Node
Worker Node

Master Node Driver

Worker Node

YARN
Spark-submit
(Cluster Mode)
Worker Node

Executor-1 Executor-2

Example: 850MB 850MB 850MB 850MB

1C 1C 1C 1C

Note:.To get the better job performance in spark, researchers have found that we can take 2 to 5 maximum core for each executor
16
Example-2: What happens when a user submits the spark application

spark-submit \
--class [Link] \
--master yarn \ 2GB-300MB(Reserved memory) =1700 MB per executor
--deploy-mode cluster \
--driver-memory 2g \
--num-executors 2 \
--executor-memory 2g \
--executor-cores 2 \
/path/to/your/spark-application_2.[Link]
Worker Node
Worker Node

Master Node Driver

Worker Node

YARN
Spark-submit
(Cluster Mode)
Worker Node

Executor-1 Executor-2

In queue

Note:.To get the better job performance in spark, researchers have found that we can take 2 to 5 maximum core for each executor

17
 To process 25GB data in spark
a. How many CPU cores are required?
b. How many executors are required?
c. How much each executor memory is required?
d. What is the total memory required?

Scenario-1: If all expected tasks runs parallelly at a time

How many executor CPU cores are required to process 25 GB data? How much each executor memory is required to process 25 GB data?

25GB = 25*1024 MB = 25600 MB CPU cores for each executor = 4

Number of Partitions = 25600 MB/128 MB = 200
Memory for each executor = 4*512 MB = 2 GB
Number of CPU Cores = Number of Partitions = 200

How many executors are required to process 25 GB data? What is the total memory required to process 25 GB data?

Total number of executor = 50

Avg CPU cores for each executor = 4 Memory for each executor = 2 GB

Total number of executor = 200/4 = 50 Total Memory for all the executor = 50*2 GB = 100 GB

Note:
➔ By default, Spark creates one partition for each block of the file (blocks being 128MB by default in HDFS), but you can also ask for a higher number of partitions by passing a larger value.
➔ To get the better job performance in spark, researchers have found that we can take 2 to 5 maximum core for each executor
➔ Expected memory for each core = Minimum 4 x (default partition size)=4*128 MB=512 MB
➔ Executor memory is not less than 1.5 times of spark reserved memory (Single core executor memory should not be less than 450MB)
18
 To process 25GB data in spark
a. How many CPU cores are required?
b. How many executors are required?
c. How much each executor memory is required?
d. What is the total memory required?

Scenario-2: If 50% of expected tasks runs parallelly at a time

How many executor CPU cores are required to process 25 GB data? How much each executor memory is required to process 25 GB data?

25GB = 25*1024 MB = 25600 MB CPU cores for each executor = 4

Number of Partitions = 25600 MB/128 MB = 200
Number of CPU Cores = Number of Partitions = 200 (To process all at a time) Memory for each executor = 4*512 MB = 2 GB

Number of CPU Cores required to process 100 partitions = 100 (To process 50% of partitions at a time)

How many executors are required to process 25 GB data? What is the total memory required to process 25 GB data?

Avg CPU cores for each executor = 4 Total number of executor = 25

Memory for each executor = 2 GB
Total number of executor = 100/4 = 25
Total Memory for all the executor = 25*2 GB = 50 GB

Note:
➔ By default, Spark creates one partition for each block of the file (blocks being 128MB by default in HDFS), but you can also ask for a higher number of partitions by passing a larger value.
➔ To get the better job performance in spark, researchers have found that we can take 2 to 5 maximum core for each executor
➔ Expected memory for each core = Minimum 4 x (default partition size)=4*128 MB=512 MB
➔ Executor memory is not less than 1.5 times of spark reserved memory (Single core executor memory should not be less than 450MB)
19
Thank You