Controlling Executors and Cores in Spark Applications

In this post, I will cover the core concepts which govern the execution model of Spark. I will talk about the different components, how they interact with each other and what happens when you fire a query. I will also take few examples to illustrate how Spark configs change these behaviours.

Let's talk about the architecture first. Each Spark application (instance of SparkContext) starts with a Driver program. The Driver talks to the Cluster Manager( YARN, Mesos, Kubernetes etc.) to demand the resources it needs. Executors (being one such resource) are processes which run computations and stored data for applications. Each application gets its own executor processes on Worker Nodes. Effectively, it means that data can't be shared across different applications without writing to a storage layer. Also, executors typically run for the lifetime of a Spark application and run multiple tasks over its lifetime (parallelly and sequentially).


What happens when you submit a job?
When you submit any job to Spark, following events occur:

1. SparkContext connects to the cluster manager
2. Spark acquires Executors on worker nodes. By default, number of executors is equal to the number of worker nodes available
3. Spark sends the application code to executors. This code might be JARs or Python files passed to the SparkContext
4. SparkContext sends the tasks to the executors to run

Optimizing Spark?
Now then, how do we control these components to extract the best out of our Spark cluster? There are some Spark configurations which can be used to make sure that our cluster resources are being used in an optimal way by our applications. Some of the important configs are:

--executor-cores : Number of concurrent tasks an executor can run. In other words, this is the number of cores per executor. It also means that if you have assigned only one core per executor, you will not be able to run any parallel tasks in the JVM
--num-executors: Though, by default, the number of executors is equal to the number of worker nodes, it can be changed using this property. Keep in mind that this property controls the number of executors requested not assigned. Assignment is successful only if there are enough cores available on worker nodes (depending on --executor-cores value)
--total-executor-cores: Maximum number of executor cores allowed per application

Let's try to understand these concepts using some examples:

Example 1: A standalone cluster with 5 worker nodes (each node having 8 cores) When i start an application with default settings. How many executors will get assigned?
Answer: Spark will greedily acquire as many cores and executors as are offered by the scheduler. So in the end you will get 5 executors with 8 cores each.
Example 2 Same cluster config as example 1, but I run an application with the following settings --executor-cores 10 --total-executor-cores 10.
Answer: Spark won't be able to allocate as many cores as requested in a single worker (as each worker only has 8 cores), hence no executors will be launched.
Another excellent example comes from the Cloudera blog

To hopefully make all of this a little more concrete, here’s a worked example of configuring a Spark app to use as much of the cluster as possible: Imagine a cluster with six nodes running NodeManagers, each equipped with 16 cores and 64GB of memory. The NodeManager capacities, yarn.nodemanager.resource.memory-mb and yarn.nodemanager.resource.cpu-vcores, should probably be set to 63 * 1024 = 64512 (megabytes) and 15 respectively. We avoid allocating 100% of the resources to YARN containers because the node needs some resources to run the OS and Hadoop daemons. In this case, we leave a gigabyte and a core for these system processes. Cloudera Manager helps by accounting for these and configuring these YARN properties automatically.
The likely first impulse would be to use --num-executors 6 --executor-cores 15 --executor-memory 63G. However, this is the wrong approach because:
63GB + the executor memory overhead won’t fit within the 63GB capacity of the NodeManagers. The application master will take up a core on one of the nodes, meaning that there won’t be room for a 15-core executor on that node. 15 cores per executor can lead to bad HDFS I/O throughput.
A better option would be to use --num-executors 17 --executor-cores 5 --executor-memory 19G. Why?
This config results in three executors on all nodes except for the one with the AM, which will have two executors. --executor-memory was derived as (63/3 executors per node) = 21. 21 * 0.07 = 1.47. 21 – 1.47 ~ 19. 

Comments

Post a Comment

Popular posts from this blog

Uber Data Model

Image
In this post I will try to come up with a data model which can serve the requirements of ride sharing companies like Uber, Lyft, Ola etc. We will approach the problem as an interview and see how we can come up with a feasible data model by answering important questions. Important Entities The first step towards building a data model is to identify important actors/ entities involved in the process. In our case, if we think about our interaction with taxi apps, we can identify important entities involved. The user (i.e. Rider) is one such entity, so is the Driver/ Partner . Once we open the app, we try to book a trip by finding a suitable taxi/ cab from a particular location to another . After the trip gets finished, the app collects the payment and we are done . Ideally, the flow continues to reviews/ ratings, helpcenter in case of issues etc. but for this post we will only consider scenarios till the ride gets finished. So, to summarize, we have the following key entities;
Read more

Data Engineer Interview Questions: SQL

Image
In this post, I will try to share some actual questions asked by top companies for Data Engineer positions. A lot of these companies will cover data modelling as one of the rounds and will use the data model for the next round based on SQL queries. Q1: Find the number of drivers available for rides in any area at any given point of time. Q2: Do you consider Driver and Rider as separate entities? Why or why not? Q3: Give me all passenger names who used the app for only airport rides. Q4: How will you decide where to apply surge pricing? Q5: How will you calculate wait times for rides? Q6: A driver can ride multiple cars, how will you find out who is driving which car at any moment? Q7: Find out Rank without using any function. Q8: How will you delete duplicates from a table? Q9: How will you find percentile? Q10:  You have 3 tables, user_dim (user_id, account_id), account_dim (account_id, paying_customer), and dload_facts (date, user_id, and downloads), find the ave
Read more

Cracking Data Engineering Interviews

Of late, a lot of people have asked me for tips on how to crack Data Engineering interviews at FAANG (Facebook, Amazon, Apple, Netflix, Google) or similar companies. I have not worked at all of these companies so I can't share tips which will necessarily apply for all of them but I will share tips which can be generalized for most of the big companies. First of all, the field of Data Engineering has expanded a lot in the last few years and has become one of the core functions of any big technology company. The obvious reason for this expansion is the amount of data being generated by devices and data-centric economy of the internet age. Each company is focussed on making the best use of data owned by them by making data driven decisions. If you compare this to the Data Engineering roles which used to exist a decade back, you will see a huge change. In the past, Data Engineering was invariably focussed on Databases and SQL. Even now, these two form some part of most Data Engin
Read more

Hive Challenges: Bucketing, Bloom Filters and More

In the second post of this series, we will learn about few more aspects of table design in Hive. Now that you know about partitioning challenges , you will be able to appreciate these features which will help you to further tune your Hive tables. Bucketing In addition to Partitioning the tables, you can enable another layer of bucketing of data based on some attribute value by using the Clustering method. Clustering can be used with partitioned or non-partitioned hive tables. One particular use case where Clustering becomes useful when your partitions might have unequal number of records (e.g. users logging in per country, US partition might be a lot bigger than New Zealand). In such cases, you can define the number of buckets and the clustered by field (like user Id), so that all the buckets have equal records. In partitioning each partition gets a directory while in Clustering, each bucket gets a file. Records with the same bucketed column will always be stored in the same bucke
Read more

UBER Data Architecture

Image
In this post, I will try to cover the data architecture Uber has built to support their big data applications. This should be applicable for other ride hailing apps as well. There are multiple modules at play here so I will try to give a brief overview along with a detailed discussion on the data architecture.  An aeroplane view of the problem tells us that effectively we are trying to solve a demand vs. supply problem. All the Drivers active at any point of time constitute the supply while all the cab requesters (Riders) form the demand. What Uber tries to do is to have the best matching between demand and supply "at any point at any moment". I have highlighted the location and temporal conditions as they are critical for the success of what Uber tries to do. Let's get started to see how Uber comes around these challenges by creating a scalable data architecture. Every Driver active on Uber keeps sending his location data to the server (e.g. every 5 seconds). T
Read more