Advent of 2020, Day 29 – Performance tuning for Apache Spark

This article is originally published at https://tomaztsql.wordpress.com

Series of Azure Databricks posts:

• Dec 01: What is Azure Databricks
• Dec 02: How to get started with Azure Databricks
• Dec 03: Getting to know the workspace and Azure Databricks platform
• Dec 04: Creating your first Azure Databricks cluster
• Dec 05: Understanding Azure Databricks cluster architecture, workers, drivers and jobs
• Dec 06: Importing and storing data to Azure Databricks
• Dec 07: Starting with Databricks notebooks and loading data to DBFS
• Dec 08: Using Databricks CLI and DBFS CLI for file upload
• Dec 09: Connect to Azure Blob storage using Notebooks in Azure Databricks
• Dec 10: Using Azure Databricks Notebooks with SQL for Data engineering tasks
• Dec 11: Using Azure Databricks Notebooks with R Language for data analytics
• Dec 12: Using Azure Databricks Notebooks with Python Language for data analytics
• Dec 13: Using Python Databricks Koalas with Azure Databricks
• Dec 14: From configuration to execution of Databricks jobs
• Dec 15: Databricks Spark UI, Event Logs, Driver logs and Metrics
• Dec 16: Databricks experiments, models and MLFlow
• Dec 17: End-to-End Machine learning project in Azure Databricks
• Dec 18: Using Azure Data Factory with Azure Databricks
• Dec 19: Using Azure Data Factory with Azure Databricks for merging CSV files
• Dec 20: Orchestrating multiple notebooks with Azure Databricks
• Dec 21: Using Scala with Spark Core API in Azure Databricks
• Dec 22: Using Spark SQL and DataFrames in Azure Databricks
• Dec 23: Using Spark Streaming in Azure Databricks
• Dec 24: Using Spark MLlib for Machine Learning in Azure Databricks
• Dec 25: Using Spark GraphFrames in Azure Databricks
• Dec 26: Connecting Azure Machine Learning Services Workspace and Azure Databricks
• Dec 27: Connecting Azure Databricks with on premise environment
• Dec 28: Infrastructure as Code and how to automate, script and deploy Azure Databricks with Powershell

Yesterday we looked into powershell automation for Azure Databricks and how one can create, updae or remove the Workspace, resource group and VNet, using deployment templates and parameters.

Today we will address the issues with Spark performance. We have talked several times about different languages available in Spark.

There are indirect and direct performance improvements that can leverage and make your Spark run faster.

1.Choice of Languages

Java versus Scala versus R versus Python (versus HiveSQL)? There is no correct or wrong answer to this choice, but there are some important differences worth mentioning. If you are running single-node machine learning the SparkR is a best option, since it has a massive machine learning ecosystem, and it has a lot of optimised algorithms that can handle this.

If you are running ETL job, Spark and combination of another language (R, Python, Scala) will yield all great results. Spark’s Structured API are consistent in terms of speed and stability across all the languages, so there should be almost none differences. But things get much more interesting when there are UDF (user defined functions) that can not be directly created in Structured API’s. In this case, both R nor Python might not be a good idea, simply because the way Structured API manifests and transforms as RDD. In general, Python makes better choice over R when writing UDF, but probably the best way would be to write UDF in Scala (or Java), making these language jumps easier for the API interpreter.

2.Choice of data presentation

DataFrame versus Datasets versus SQL versus RDD is another choice, yet it is fairly easy. DataFrames, Datasets and SQL objects are all equal in performance and stability (at least from Spar 2.3 and above), meaning that if you are using DataFrames in any language, performance will be the same. Again, when writing custom objects of functions (UDF), there will be some performance degradation with both R or Python, so switching to Scala or Java might be a optimisation.

Rule of thumb is, stick to DataFrames. If you go a layer down to RDD, Spark will make better optimisation and use of it than you will. Spark optimisation engine will write better RDD code than you do and with certainly less effort. And doing so, you might also loose additional Spark optimisation with new releases.

When using RDD, try and use Scala or Java. If this is not possible, and you will be using Python or R extensively, try to use it as little as possible with RDDs. And convert to DataFrames as quickly as possible. Again, if your Spark code, application or data engineering task is not compute intensive, it should be fine, otherwise remember to use Scala or Java or convert to DataFrames. Both Python and R does not handle serialisation of RDD files optimally and runs a lot of data to and from Python or R engine, causing a lot of data movement, traffic and potentially making RDD unstable and making poor performance.

3. Data Storage

Storing data effectively is relevant when data will be read multiple times. If data will be accessed many times, either from different users in organisation or from a single user, all making data analysis, make sure to store it for effective reads. Choosing your storage, choosing the data formats and data partitioning is important.

With numerous file format available, there are some key differences. If you want to optimise your Spark job, data should be stored in best possible format for this. In general, always favour structured, binary types to store your data, especially when you are doing frequent-accessing. Although CSV files look well formatted, they are obnoxiously sparse, can have “edge” cases (missing line breaks, or other delimiters) are painfully slow to parse and hard to partition. Same logic applies to txt and xml formats. Avro are JSON orientated and also sparse and I am not going to even talk about XML format. Spark works best with Apache Parquet stored data. Parquet format stores data in a binary files in column-orientated storage, and also track some statistics of the files, making it possible to skip files not needed for query.

4.Table partitioning and bucketing

Table partitioning is referring to storing files in separate directories based on a partition key (e.g.: date of purchase, VAT number) such as a date field in data stored in these directories. Partitioning will help Spark skip files that are not needed for end result and it will return only the data that is in the range of the key. There are potentials pitfalls to this techniques, one for sure is the size of these subdirectories and how to choose the right granularity.

Bucketing is a process of “pre-partitioning” data to allow better data joins and aggregations operations. This will improve performance, because data can be consistently distributed across partitions as opposed all being in one partition. So if you are repeating a particular query that is joins are frequently performed on a column immediately after read, you can use bucketing to assure that data is well partitioned in accordance with those values. This will prevent shuffle before join and speed up data access.

5.Parallelism

Splittable data formats make Spark job easier to run in parallel. A ZIP or a TAR file can not be split, which means that even if you have 10 files in a ZIP file and 10 cores, only one core can read in that data, because Spark can not parallelise across ZIP file. But using GZIP, BZIP2 or LZ4 are generally splittable if (and only if) they are written by a parallel processing framework like Spark or Hadoop.

In general, Spark will work best when there are two to three tasks per CPU core in your cluster when working especially with large (big) data. You can also tune the spark.default.parallelism property.

6.Number of files

With numerous small files you will for sure pay a price for listing and fetching all the data. There is no golden rule on number of files and the size of the files per directory. But there are some directions. Multiple small files will is going to make a schedule worked harder to locate the data and launch all the read tasks. This can increase not only disk I/O but also network traffic. On the other spectrum, having fewer and larger files can ease the workload from scheduler, but it will make tasks longer to run. Again, a rule of thumb would be, to scope the size of the files in such way, that they contain a few tens of megabyte of data. From Spark 2.2. onward there are also possibilities to to partitioning and sizing optionally.

7. Temporary data storage

Data that will be reused constantly are great candidates for caching. Caching will place a DataFrame, Dataset, SQL table or RDD into temporary storage (either memory or disk) across the executors in your cluster. You might want to cache only dataset that will be used several times later on, but should not be hastened, because it takes also resources such as serialisation, deserialisation and storage costs. You can tell Spark to cache data by using a cache command on DataFrames or RDD’s.

Let’s put this to the test. In Azure Databricks create a new notebook: Day29_tuning and language: Python and attach the notebook to your cluster. Load a sample CSV file:

%python
DF1 = spark.read.format("CSV")
       .option("inferSchema", "true")
       .option("header","true")
       .load("dbfs/databricks-datasets/COVID/covid-19-data/us-states.csv")

The bigger the files, the more evident the difference will be. Create some aggregations:

DF2 = DF1.groupby("state").count().collect()
DF3 = DF1.groupby("date").count().collect()
DF4 = DF1.groupby("cases").count().collect()

After you have tracked the timing, now, let’s cache the DF1:

DF1.cache()
DF1.count()

And rerun the previous command:

DF2 = DF1.groupby("state").count().collect()
DF3 = DF1.groupby("date").count().collect()
DF4 = DF1.groupby("cases").count().collect()

And you should see the difference in results. As mentioned before, the bigger the dataset, the bigger would be time gained back when caching data.

We have touched today couple of performance tuning points and what approach one should take, to improve the work of Spark in Azure Databricks. These are probably the most frequent performance tunings and relatively easy to adjust.

Tomorrow we will look further into Apache Spark.

Complete set of code and the Notebook is available at the Github repository.

Happy Coding and Stay Healthy!

Thanks for visiting r-craft.org
This article is originally published at https://tomaztsql.wordpress.com
Please visit source website for post related comments.