Apache Spark performance tuning

Good understanding of data is where effective performance tuning boils down to then. Many of developers, especially pure python or java programmers try to apply some approaches (bucketing, copartiioning, setting system parameters, compressioning) without good grasp of underlying data. That’s why before tackling tasks we advise you to allocate some time for learning basic spark techniques

First, I should note that we could start our investigation with Dr Elephant framework. It allows you to see the main bottlenecks of your workflow and possible ways for resolving.

Besides it, we can consider cluster profiling for more tricky cases (low-level). You can see a good example of it here https://www.youtube.com/watch?v=Yh9KxQ3fKy0

Cluster level

In the beginning let’s look into cluster level. When you run your application, in spark UI you can see memory used parameter (20GB) and memory available (200Gb). If you have such difference, it means that cluster utilization is not effective and resources work in idle. It can happen when you used not splittable large file as a source. On the other hand, there can be a situation where memory used parameter (160GB) and memory available (200Gb). In this case you can ask where 40GB is? The answer is that amount was not allocated due to executor size (80GB). 200 = 80+80+40

In this regard, let’s look into executor container structure.

• Executor Memory
• Overhead (off-heap memory) = ExecutorMemory*0,1

In its turn Executor memory = JVM used memory

JVM memory

• Execution Memory and Storage memory (controlled by fraction) = Unified Memory = 0,6* ExecutorMemory
• User Memory = 0,4 * ExecutorMemory

 It should be pointed out that here can be situations when we have executor memory =16 GB, but your app (JVM) uses 6 GB MAX, so it is better to set spark.executor memory to lower value, i.e. 7 GB.

Or Allocated Unified memory can be 10GB*0,6= 6GB, but MAX JVM memory used = 7,2GB and MAX peak unified memory = 1,2GB. So it would be better to use spark.memory.fraction = 0,2GB

The can be also a case when you use two big executor (for example 10GB) with 3 cores. It will lead to spark.executor.instances =3. But your shuffle generates max 200 tasks. In this case you can either increase spark.shuffle.partitions or modify your code to have more partitions or reduce the number of executor cores.

Dr Elephant can easily help to detect such cases.

Spark API

You can work with spark using RDD, Dataframe, Datasets.

Resilient Distributed Dataset - immutable distributed collection of elements of your data, partitioned across nodes in your cluster that can be operated in parallel with a low-level API that offers transformations and actions.

It is used when you want low-level transformation and actions and control on your dataset or your data is unstructured, such as media streams or streams of text. DataFrames and Datasets are built on top of RDDs.

From production experience I can say that it is better to use dataframe api from performance optimization point of view. Under the hood spark optimizes query in the better way by default if you use high-level API,

Like an RDD, a DataFrame is an immutable distributed collection of data.

Unlike an RDD, data is organized into named columns, like a table in a relational database.  DataFrame allows developers to impose a structure onto a distributed collection of data, allowing higher-level abstraction. Both scala and python API are approximately the same efficient if you use Dataframe.

The only difference that under the hood python function will be serialized and deserialized. It causes some mess in spark UI: difficult to read DAG diagram (it is not straightforward). In addition UDF can be very slow if you use Python, because a new function can be opaque for Spark, regular function like map, filter are already optimized under the hood. Since SQL functions are relatively simple and are not designed for complex tasks it is pretty much impossible compensate the cost of repeated serialization, deserialization and data movement between Python interpreter and JVM. Unlike UDFs, Spark SQL functions operate directly on JVM and typically are well integrated with both Catalyst and Tungsten. It means these can be optimized in the execution plan and most of the time can benefit from codgen and other Tungsten optimizations. Moreover, these can operate on data in its "native" representation.

Dataset is a strongly-typed API available in Scala and Java. The main benefit is static-typing and runtime type-safety. Another benefit is high-level abstraction and custom view into structured and semi-structured data. It is built on the top of untyped DataFrame (so dataframe is even faster and suitable for interactive analysis). Dataset is optimized for data engineering tasks

Because DataFrame and Dataset APIs are built on top of the Spark SQL engine, it uses Catalyst to generate an optimized logical and physical query plan.

Types of dependencies in spark

There are two kind of operations:

Transformations have RDD input and RDD output, lazy evaluated. They get execute when we call an action. They are not executed immediately (groupByKey(), flatMap)

Actions execute transformations (count).

There are narrow and wide transformations. Narrow ones take one partition in RDD and return one partition. Narrow contains map and filter + sample + union

Wide ones take one partition in RDD and can return more than one. Shuffling takes place in this place. It impacts performance significantly. There are Coalesce, Repartition, Join, Distinct, Reducebykey, groupbykey

Types of joins in spark

Now it would be insightful to look into types of joins in spark. Physical plan will help us to see which type of joins

Broadcast Hash Join is the most performant join strategy. It ships smaller table to all nodes and then stream it to other tables. But it is applicable to a small set of scenarios (is not supported for full outer join.). To use it you should usually use hint.

Shuffle Hash Join & Sort Merge Join are the true work-horses of Spark SQL; a majority of the use-cases involving joins you will encounter in Spark SQL will have a physical plan using either of these strategies.

Shuffle Hash Join means shuffling both tables, hash smaller one and stream to bigger one. If the average size of a single partition is small enough to build a hash table (any partition of the build side could fit in memory).  The build side is much smaller than stream side, the building hash table on smaller side should be faster than sorting the bigger side.

Sort Merge Join is optimal under certain conditions. For example, the Datasets are distributed evenly on the join columns. if the matching join keys are sortable. Sort Merge Join which is faster (in some cases) and more robust. It’s definitely more robust as Shuffled Hash Join requires the hashed table to fit in memory, counter to Sort Merge Join which can spill to disk.

Cartesian products can be a very expensive operation. Even with as little as 6,000 products and 100,000 customers, the output will be 60 million records (6K x 100K = 600M)

Thus, there pretty lot of nuances of perfomance tuning in Spark. In the articles we will continue enlightening them