Best Practices for Spark Thrift Server on YARN

This article covers best practices for the following topics:
1. Caching
2. Broadcast Lookup Tables
3. Parallelism
4. Reduce Memory Overhead
5. Serialization and Compression
6. Performance Improvements
7. Conclusions

Caching

Pin tables in memory using "cache table <table-name>" statement.

This optimization improves filtering, aggregation and joins. This will load tables in Spark nodes' memory and improve future queries to the table. Make sure the size of the tables to load (how to check table size and total cluster storage size here), is less than the storage memory (that is, check that the summation of all tables to be cached is less than ( (spark.executor.memory - 300MB) * spark.memory.fraction * spark.memory.storageFraction )   ).
Ref: https://spark.apache.org/docs/latest/sql-programming-guide.html#caching-data-in-memory

Broadcast Lookup Tables

Set broadcast table size limit setting to be larger than the lookup table sizes but smaller than the execution memory of a given executor node. "spark.sql.autoBroadcastJoinThreshold = 10485760".

When Spark SQL plans joins, it will identify whether the tables to join are smaller than the limit spark.sql.autoBroadcastJoinThreshold, and if so, transfer the tables to all the executors. This allows for faster joins. Make sure the value is not set to larger than the execution memory each executor has. Please check here the execution memory each executor has.
Ref: https://spark.apache.org/docs/latest/sql-programming-guide.html#other-configuration-options

Parallelism

Increase number of cores

In most large queries, the more a query can be parallelized, the faster it will execute. We can increase the parallelization of a query by increasing the total number of cores. Do this by increasing the number of instances (spark.executor.instances) or increase the number of cores per instance (spark.executor.cores) or both. After doubling the amount of cores we saw performance was improved by about ~50%. We found that the larger the query was, the more impact increasing the amount of cores would have on the performance. In general we should maximize the total number of cores defined as
Total number of cores = spark.executor.cores * spark.executor.instances
spark.executor.cores (default value is 1 in YARN mode and all available nodes in the worker node in Mesos or Standalone mode): the number of cores each executor instance will have.
spark.executor.instances (default value is 2) : the number of executor instances. Note that this property is incompatible with spark.dynamicAllocation.enabled. If both spark.dynamicAllocation.enabled and spark.executor.instances are specified, dynamic allocation is turned off and the specified number of spark.executor.instances is used.
Ref: https://spark.apache.org/docs/1.6.1/running-on-yarn.html#spark-properties
Ref: https://spark.apache.org/docs/latest/configuration.html#execution-behavior

Set number of partitions to the smallest prime number larger or equal to the number of cores in the cluster: "set spark.sql.shuffle.partition=13"

This improves joins performance. When doing joins, partitions or aggregations, the resulting tables will be partitioned according with this value. It is a measure of parallelism, it defines how many reducers will exist after the  joins or aggregations. By default it is 200. If the number of partitions is lower than the number of cores in the cluster, we might not utilize all the cores in the cluster. Therefore is good idea to set the value to the smallest prime number larger or equal to the number of cores in the cluster. We set the value to be a prime number in order to evenly spread the load between all reducers (please check forth reference below about prime numbers and hash functions).
Ref: https://spark.apache.org/docs/latest/sql-programming-guide.html#other-configuration-options
Ref: https://issues.apache.org/jira/browse/SPARK-18479
Ref: https://spark.apache.org/docs/latest/tuning.html#level-of-parallelism
Ref: http://srinvis.blogspot.com/2006/07/hash-table-lengths-and-prime-numbers.html, also in CLRS Chapter 31st, theorem 31.20.

Repartition source table for faster table processing.

We see performance improvements if we repartition the table across an evenly distributed attribute (like date). What this repartition achieves is to make the partitions about same size. Each partition is handled by only one core. And once partitions are about the same size, the executors will finish processing the partitions at about the same time. When the partitions were not the same size, some executors finished faster or slower depending on the partition size.
Example:
A table "gamereviews" is partitioned in HDFS in 2 parts. When reading HDFS files, spark assigns one task per HDFS partition. Therefore, when querying table "gamereviews", two tasks are created to process the two HDFS partitions. Each task is executed by a different core in parallel.
>select std(score) from gamereviews;

Ref: https://dzone.com/articles/optimize-spark-with-distribute-by-cluster-by

Reduce Memory Overhead

Turn on compression for in memory tables "spark.sql.inMemoryColumnarStorage.compressed = true", "spark.sql.inMemoryColumnarStorage.batchSize =10000"

Make sure the compression of in memory tables is enabled, this will allow us to use memory efficiently. Also, without in-memory table compression, the cache command will not be available. The spark.sql.inMemoryColumnarStorage.batchSize controls the size of batches for columnar caching. Larger batch size can improve memory utilization and compression but risks out of memory when caching and requesting a new batch. Setting default 1000 is a good start. Starting Spark 1.2.0 the default values are spark.sql.inMemoryColumnarStorage.compressed=true, spark.sql.inMemoryColumnarStorage.batchSize =10000.
Ref: https://spark.apache.org/docs/latest/sql-programming-guide.html#caching-data-in-memory 

If Spark version is older than 1.5, turn on tungsten. "set spark.sql.tungsten.enabled=true"

Tungsten is an optimization done to the memory management in Spark that allows for efficient use of the memory. In short, the tungsten removes many of the objects memory footprint and leave only the actual data in memory. It is the only memory management after spark 1.5, but prior to 1.5 is good idea to turn it on.
Ref: https://community.hortonworks.com/articles/72502/what-is-tungsten-for-apache-spark.html
Ref:http://spark.apache.org/releases/spark-release-1-5-0.html

Turn on the "spark.memory.offHeap.enabled=true", "spark.memory.offHeap.size=1g"

Spark makes use of the Native OS memory (outside of JVM heap) if the spark.memory.offHeap.enabled and spark.memory.offHeap.size are set. The benefit of using Native OS memory is offloading work for garbage collector because it won't collect any memory allocated outside of the JVM heap. Please notice that both JVM heap and Native OS memory compete for the same limited amount of physical memory and therefore when assigning memory through spark.memory.offHeap.size, consider decreasing in similar proportion the spark.executor.memory to avoid out of memory errors.
Ref: https://spark.apache.org/docs/latest/configuration.html#memory-management
Ref: https://dzone.com/articles/heap-vs-heap-memory-usage
 

Serialization and Compression

Use Kryo serializer

Serialization plays an important role in the performance of any distributed application. Formats that are slow to serialize objects into, or consume a large number of bytes, will greatly slow down the computation. Often, this will be the first thing you should tune to optimize a Spark application. Spark can also use the Kryo library (version 2) to serialize objects more quickly. Kryo is significantly faster and more compact than Java serialization (often as much as 10x), but does not support all Serializable types and requires you to register the classes you’ll use in the program in advance for best performance. Set it using "spark.serializer" = "org.apache.spark.serializer.KryoSerializer"
Ref: https://spark.apache.org/docs/latest/tuning.html#data-serialization
 

Use snappy compression codec

The codec used to compress internal data such as RDD partitions, broadcast variables and shuffle outputs. By default, Spark provides three codecs: lz4, lzf, and snappy. We recommend using snappy based on this comparison: http://blog.erdemagaoglu.com/post/4605524309/lzo-vs-snappy-vs-lzf-vs-zlib-a-comparison-of 
"spark.io.compression.codec"="snappy"
 

Use ORC format instead of TEXT format

When tables are stored as TEXT format, they occupy much more space in HDFS. After we convert the partition tables to ORC, they have faster response times for large tables.
 

 

Performance improvements

Below two experiments: both test different datasets and different settings. We add them both here as evidence of the effect of the different tuning options:

Experiment 1: cache table, compression, broadcast join, optimized partition and offheap memory

Below find the experiments table with the actual times. The environment used for this test has the following parameters:
Hortonworks 2.6
Spark 1.6.3
3 master nodes each with  16 MB of RAM and 4 vCPU (AWS EC2 m4.xlarge)
6 slave node each with 16 MB of RAM and 4 vCPU (AWS EC2 m4.xlarge)
 
Yarn relevant parameters constants throughout all cases.
yarn.nodemanager.resource.memory-mb = 12288
yarn.nodemanager.pmem-check-enabled = true
yarn.scheduler.maximum-allocation-mb = 12288
 
Spark fixed parameters:
spark.executor.cores=2
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.initialExecutors=6
spark.dynamicAllocation.maxExecutors=6
spark.dynamicAllocation.minExecutors=6
spark.memory.fraction=0.75
spark.memory.storageFraction=0.75
spark.yarn.executor.memoryOverhead=2048
 
 
Three cases were tested: Filter, Aggregation and Join.
 
Test data: 
 
Performance tests

Experiment 2: ORC, partition and cache

We ran the following query against two different sized tables and then recorded the amount of time it took to execute: select a , b, c, sum(d) from A  where date between 'first date' and 'second date' group by a, b, c. All times below in seconds. 

Conclusions

The caching tables optimization enhances all operations: filtering, aggregation and joining. 
Auto broadcasting optimization enhances join performance by reducing the execution time by more than half (from 56 secs to 19 secs).
We see join performance improvements by tuning the number of partitions.
On the other hand we see that turning on the off-heap memory consumption does not translate to tangible improvements.