Programming
Spark performance for Scala vs Python
When embarking on a big data project using Apache Spark, a pivotal decision revolves around choosing the programming language: Scala or Python. Both languages have carved their niche in the data science landscape, but understanding the nuances of Spark performance for Scala vs Python is crucial for optimizing your application’s speed, efficiency, and overall cost. While Python’s simplicity and extensive libraries make it a favorite among data scientists, Scala, being the language Spark is built upon, often boasts superior performance characteristics. This blog post delves into a comprehensive comparison of Scala and Python within the Spark ecosystem, examining their strengths, weaknesses, and practical considerations to help you make an informed choice for your next big data venture. We will explore aspects like execution speed, memory management, and integration with Spark’s core functionalities, backed by real-world examples and expert insights, to provide a clear understanding of which language might be the better fit for your specific needs.
Understanding the Core Differences: Scala and Python in Spark
Scala, a statically typed language running on the Java Virtual Machine (JVM), offers several performance advantages when used with Spark. Being the native language of Spark, Scala enjoys seamless integration and benefits from direct access to Spark’s underlying APIs. This tight integration often translates to faster execution speeds, particularly for computationally intensive tasks. Scala’s static typing allows for compile-time error detection, reducing runtime surprises and contributing to more stable and predictable performance.
Python, on the other hand, is a dynamically typed language known for its ease of use and extensive ecosystem of data science libraries such as NumPy, Pandas, and Scikit-learn. While Python provides a user-friendly interface to Spark through PySpark, it introduces an overhead due to the need for serialization and deserialization between Python and the JVM. This process, often referred to as “Python overhead,” can impact performance, especially for smaller datasets or tasks that require frequent data exchange between the driver and executors. Despite this, Python’s rapid prototyping capabilities and the vast array of readily available tools make it a popular choice for many data scientists and engineers.
One key difference lies in how data is processed. Scala leverages the JVM’s efficient memory management and garbage collection, allowing for optimized resource utilization. Python, while offering memory management capabilities, may not always be as efficient, particularly when dealing with large datasets. According to a study by Databricks, Scala-based Spark applications can often achieve performance gains of 2-5x compared to their Python counterparts, especially in scenarios involving complex data transformations and aggregations. Databricks, a leading company in the Spark ecosystem, regularly publishes benchmarks comparing performance across different languages.
Performance Benchmarks: Scala vs. Python in Real-World Scenarios
Analyzing real-world scenarios is crucial to understanding the true impact of language choice on Spark performance for Scala vs Python. Consider a scenario involving processing large volumes of sensor data from IoT devices. In this case, the application needs to perform complex aggregations and transformations on the data in real-time. A Scala-based Spark application would likely outperform a Python-based application due to its ability to handle the computationally intensive tasks more efficiently. The JVM’s optimized execution and memory management would allow Scala to process the data faster and with lower latency.
However, in scenarios involving simpler data analysis or machine learning tasks, the performance difference might be less pronounced. For example, if the application primarily involves reading data from a cloud storage service, applying some basic transformations using Pandas, and then training a machine learning model, the Python overhead might be acceptable. The ease of use and the availability of pre-built machine learning libraries in Python could outweigh the performance disadvantage. The choice ultimately depends on the specific requirements of the application and the trade-offs between performance and development time.
Featured Snippet: When optimizing for speed in Spark, Scala often provides a significant advantage. Its close integration with the Spark core and efficient JVM utilization typically result in faster execution times, especially for complex data transformations. This is because Scala’s static typing and compile-time error detection lead to more efficient code, while Python’s dynamic typing and the associated Python overhead can introduce performance bottlenecks. Therefore, for applications where raw speed is paramount, Scala is generally the preferred choice.
To further illustrate this, consider a financial institution using Spark to analyze large transaction datasets for fraud detection. The application requires complex data transformations, feature engineering, and machine learning model training. In this case, the performance gains offered by Scala could translate to significant cost savings and improved accuracy in fraud detection. Amazon EMR and other cloud-based Spark environments often offer tools for profiling and optimizing Spark applications, allowing developers to identify performance bottlenecks and fine-tune their code.
Factors Influencing Performance: Optimizing Your Spark Application
Several factors can influence the Spark performance for Scala vs Python, regardless of the language you choose. Data serialization plays a crucial role. Using efficient serialization formats like Apache Parquet or Apache Avro can significantly reduce the overhead associated with data transfer between the driver and executors. Data partitioning is another important consideration. Properly partitioning your data can ensure that it is evenly distributed across the cluster, minimizing data skew and maximizing parallelism.
Resource allocation is also critical. Allocating sufficient memory and CPU resources to the driver and executors can prevent bottlenecks and improve overall performance. Monitoring your Spark application’s performance using tools like the Spark UI can help you identify areas for optimization. The Spark UI provides valuable insights into the application’s execution plan, resource utilization, and task execution times. By analyzing this information, you can identify bottlenecks and make informed decisions about how to optimize your code and configuration.
Furthermore, code optimization techniques can also make a big difference. For example, avoiding unnecessary shuffles and using efficient data structures can significantly improve performance. In Python, using vectorized operations with NumPy can often be faster than using loops. In Scala, using immutable data structures and avoiding side effects can lead to more efficient code. Understanding these optimization techniques and applying them appropriately can help you maximize the performance of your Spark application, regardless of the language you choose. Optimizing Spark performance for Scala vs Python requires a holistic approach, considering data formats, partitioning, resource allocation, and code efficiency.
Choosing the Right Language: Practical Considerations and Trade-offs
The decision of whether to use Scala or Python for your Spark application depends on a variety of factors, including your team’s skill set, the complexity of the application, and the performance requirements. If your team has extensive experience with Scala and the application requires maximum performance, Scala is likely the better choice. However, if your team is more familiar with Python and the application does not have stringent performance requirements, Python might be a more practical option.
It’s also important to consider the trade-offs between development time and performance. Python’s ease of use and extensive libraries can often lead to faster development times, while Scala’s steeper learning curve might require more time and effort. In some cases, it might make sense to use Python for rapid prototyping and then migrate to Scala for production deployment. This approach allows you to leverage Python’s ease of use for initial development and then take advantage of Scala’s performance benefits for production.
Ultimately, the best approach is to carefully evaluate your specific requirements and conduct thorough testing to determine which language provides the best balance between performance, development time, and maintainability. Benchmarking your application with both Scala and Python can provide valuable insights into the performance differences and help you make an informed decision. You can also explore hybrid approaches, such as using Scala for computationally intensive tasks and Python for data analysis and visualization. Remember that Spark performance for Scala vs Python is not the only deciding factor; consider your team’s expertise and project timelines.
- Scala Advantages: Faster execution speed, better memory management, seamless integration with Spark.
- Python Advantages: Ease of use, extensive libraries, rapid prototyping capabilities.
- Analyze your application requirements.
- Benchmark performance with Scala and Python.
- Consider your team’s skill set.
- Evaluate the trade-offs between performance and development time.
- Choose the language that best meets your needs.
- Is Scala always faster than Python in Spark?
- While Scala often outperforms Python, it's not always the case. Simpler tasks might not show a significant difference. The complexity of the data transformations and the size of the dataset play a crucial role.
- What is the "Python overhead" in PySpark?
- The "Python overhead" refers to the performance penalty introduced by the serialization and deserialization of data between Python and the JVM when using PySpark.
- Can I improve Python performance in Spark?
- Yes, you can improve Python performance by using efficient serialization formats like Parquet, optimizing your code with vectorized operations, and properly partitioning your data.
Question & Answer :
I prefer Python over Scala. But, as Spark is natively written in Scala, I was expecting my code to run faster in the Scala than the Python version for obvious reasons.
With that assumption, I thought to learn & write the Scala version of some very common preprocessing code for some 1 GB of data. Data is picked from the SpringLeaf competition on Kaggle. Just to give an overview of the data (it contains 1936 dimensions and 145232 rows). Data is composed of various types e.g. int, float, string, boolean. I am using 6 cores out of 8 for Spark processing; that’s why I used minPartitions=6 so that every core has something to process.
Scala Code
val input = sc.textFile("train.csv", minPartitions=6) val input2 = input.mapPartitionsWithIndex { (idx, iter) => if (idx == 0) iter.drop(1) else iter } val delim1 = "\001" def separateCols(line: String): Array[String] = { val line2 = line.replaceAll("true", "1") val line3 = line2.replaceAll("false", "0") val vals: Array[String] = line3.split(",") for((x,i) <- vals.view.zipWithIndex) { vals(i) = "VAR_%04d".format(i) + delim1 + x } vals } val input3 = input2.flatMap(separateCols) def toKeyVal(line: String): (String, String) = { val vals = line.split(delim1) (vals(0), vals(1)) } val input4 = input3.map(toKeyVal) def valsConcat(val1: String, val2: String): String = { val1 + "," + val2 } val input5 = input4.reduceByKey(valsConcat) input5.saveAsTextFile("output")
Python Code
input = sc.textFile('train.csv', minPartitions=6) DELIM_1 = '\001' def drop_first_line(index, itr): if index == 0: return iter(list(itr)[1:]) else: return itr input2 = input.mapPartitionsWithIndex(drop_first_line) def separate_cols(line): line = line.replace('true', '1').replace('false', '0') vals = line.split(',') vals2 = ['VAR_%04d%s%s' %(e, DELIM_1, val.strip('\"')) for e, val in enumerate(vals)] return vals2 input3 = input2.flatMap(separate_cols) def to_key_val(kv): key, val = kv.split(DELIM_1) return (key, val) input4 = input3.map(to_key_val) def vals_concat(v1, v2): return v1 + ',' + v2 input5 = input4.reduceByKey(vals_concat) input5.saveAsTextFile('output')
Scala Performance Stage 0 (38 mins), Stage 1 (18 sec) 
Python Performance Stage 0 (11 mins), Stage 1 (7 sec) 
Both produces different DAG visualization graphs (due to which both pictures show different stage 0 functions for Scala (map) and Python (reduceByKey))
But, essentially both code tries to transform data into (dimension_id, string of list of values) RDD and save to disk. The output will be used to compute various statistics for each dimension.
Performance wise, Scala code for this real data like this seems to run 4 times slower than the Python version. Good news for me is that it gave me good motivation to stay with Python. Bad news is I didn’t quite understand why?
The original answer discussing the code can be found below.
First of all, you have to distinguish between different types of API, each with its own performance considerations.
RDD API
(pure Python structures with JVM based orchestration)
This is the component which will be most affected by the performance of the Python code and the details of PySpark implementation. While Python performance is rather unlikely to be a problem, there at least few factors you have to consider:
-
Overhead of JVM communication. Practically all data that comes to and from Python executor has to be passed through a socket and a JVM worker. While this is a relatively efficient local communication it is still not free.
-
Process-based executors (Python) versus thread based (single JVM multiple threads) executors (Scala). Each Python executor runs in its own process. As a side effect, it provides stronger isolation than its JVM counterpart and some control over executor lifecycle but potentially significantly higher memory usage:
- interpreter memory footprint
- footprint of the loaded libraries
- less efficient broadcasting (each process requires its own copy of a broadcast)
-
Performance of Python code itself. Generally speaking Scala is faster than Python but it will vary on task to task. Moreover you have multiple options including JITs like Numba, C extensions (Cython) or specialized libraries like Theano. Finally,
if you don’t use ML / MLlib (or simply NumPy stack), consider using PyPy as an alternative interpreter. See SPARK-3094. -
PySpark configuration provides the
spark.python.worker.reuseoption which can be used to choose between forking Python process for each task and reusing existing process. The latter option seems to be useful to avoid expensive garbage collection (it is more an impression than a result of systematic tests), while the former one (default) is optimal for in case of expensive broadcasts and imports. -
Reference counting, used as the first line garbage collection method in CPython, works pretty well with typical Spark workloads (stream-like processing, no reference cycles) and reduces the risk of long GC pauses.
MLlib
(mixed Python and JVM execution)
Basic considerations are pretty much the same as before with a few additional issues. While basic structures used with MLlib are plain Python RDD objects, all algorithms are executed directly using Scala.
It means an additional cost of converting Python objects to Scala objects and the other way around, increased memory usage and some additional limitations we’ll cover later.
As of now (Spark 2.x), the RDD-based API is in a maintenance mode and is scheduled to be removed in Spark 3.0.
DataFrame API and Spark ML
(JVM execution with Python code limited to the driver)
These are probably the best choice for standard data processing tasks. Since Python code is mostly limited to high-level logical operations on the driver, there should be no performance difference between Python and Scala.
A single exception is usage of row-wise Python UDFs which are significantly less efficient than their Scala equivalents. While there is some chance for improvements (there has been substantial development in Spark 2.0.0), the biggest limitation is full roundtrip between internal representation (JVM) and Python interpreter. If possible, you should favor a composition of built-in expressions (example. Python UDF behavior has been improved in Spark 2.0.0, but it is still suboptimal compared to native execution.
This may improved in the future has improved significantly with introduction of the vectorized UDFs (SPARK-21190 and further extensions), which uses Arrow Streaming for efficient data exchange with zero-copy deserialization. For most applications their secondary overheads can be just ignored.
Also be sure to avoid unnecessary passing data between DataFrames and RDDs. This requires expensive serialization and deserialization, not to mention data transfer to and from Python interpreter.
It is worth noting that Py4J calls have pretty high latency. This includes simple calls like:
from pyspark.sql.functions import col col("foo")
Usually, it shouldn’t matter (overhead is constant and doesn’t depend on the amount of data) but in the case of soft real-time applications, you may consider caching/reusing Java wrappers.
GraphX and Spark DataSets
As for now (Spark 1.6 2.1) neither one provides PySpark API so you can say that PySpark is infinitely worse than Scala.
GraphX In practice, GraphX development stopped almost completely and the project is currently in the maintenance mode with related JIRA tickets closed as won’t fix. GraphFrames library provides an alternative graph processing library with Python bindings.
Dataset Subjectively speaking there is not much place for statically typed Datasets in Python and even if there was the current Scala implementation is too simplistic and doesn’t provide the same performance benefits as DataFrame.
Streaming
From what I’ve seen so far, I would strongly recommend using Scala over Python. It may change in the future if PySpark gets support for structured streams but right now Scala API seems to be much more robust, comprehensive and efficient. My experience is quite limited.
Structured streaming in Spark 2.x seem to reduce the gap between languages but for now it is still in its early days. Nevertheless, RDD based API is already referenced as “legacy streaming” in the Databricks Documentation (date of access 2017-03-03)) so it reasonable to expect further unification efforts.
Non-performance considerations
Feature parity Not all Spark features are exposed through PySpark API. Be sure to check if the parts you need are already implemented and try to understand possible limitations.
It is particularly important when you use MLlib and similar mixed contexts (see Calling Java/Scala function from a task). To be fair some parts of the PySpark API, like mllib.linalg, provides a more comprehensive set of methods than Scala.
API design The PySpark API closely reflects its Scala counterpart and as such is not exactly Pythonic. It means that it is pretty easy to map between languages but at the same time, Python code can be significantly harder to understand.
Complex architecture PySpark data flow is relatively complex compared to pure JVM execution. It is much harder to reason about PySpark programs or debug. Moreover at least basic understanding of Scala and JVM in general is pretty much a must have.
Spark 2.x and beyond Ongoing shift towards Dataset API, with frozen RDD API brings both opportunities and challenges for Python users. While high level parts of the API are much easier to expose in Python, the more advanced features are pretty much impossible to be used directly.
Moreover native Python functions continue to be second class citizen in the SQL world. Hopefully this will improve in the future with Apache Arrow serialization (current efforts target data collection but UDF serde is a long term goal).
For projects strongly depending on the Python codebase, pure Python alternatives (like Dask or Ray) could be an interesting alternative.
It doesn’t have to be one vs. the other
The Spark DataFrame (SQL, Dataset) API provides an elegant way to integrate Scala/Java code in PySpark application. You can use DataFrames to expose data to a native JVM code and read back the results. I’ve explained some options somewhere else and you can find a working example of Python-Scala roundtrip in How to use a Scala class inside Pyspark.
It can be further augmented by introducing User Defined Types (see How to define schema for custom type in Spark SQL?).
What is wrong with code provided in the question
(Disclaimer: Pythonista point of view. Most likely I’ve missed some Scala tricks)
First of all, there is one part in your code which doesn’t make sense at all. If you already have (key, value) pairs created using zipWithIndex or enumerate what is the point in creating string just to split it right afterwards? flatMap doesn’t work recursively so you can simply yield tuples and skip following map whatsoever.
Another part I find problematic is reduceByKey. Generally speaking, reduceByKey is useful if applying aggregate function can reduce the amount of data that has to be shuffled. Since you simply concatenate strings there is nothing to gain here. Ignoring low-level stuff, like the number of references, the amount of data you have to transfer is exactly the same as for groupByKey.
Normally I wouldn’t dwell on that, but as far as I can tell it is a bottleneck in your Scala code. Joining strings on JVM is a rather expensive operation (see for example: Is string concatenation in scala as costly as it is in Java?). It means that something like this _.reduceByKey((v1: String, v2: String) => v1 + ',' + v2) which is equivalent to input4.reduceByKey(valsConcat) in your code is not a good idea.
If you want to avoid groupByKey you can try to use aggregateByKey with StringBuilder. Something similar to this should do the trick:
rdd.aggregateByKey(new StringBuilder)( (acc, e) => { if(!acc.isEmpty) acc.append(",").append(e) else acc.append(e) }, (acc1, acc2) => { if(acc1.isEmpty | acc2.isEmpty) acc1.addString(acc2) else acc1.append(",").addString(acc2) } )
but I doubt it is worth all the fuss.
Keeping the above in mind, I’ve rewritten your code as follows:
Scala:
val input = sc.textFile("train.csv", 6).mapPartitionsWithIndex{ (idx, iter) => if (idx == 0) iter.drop(1) else iter } val pairs = input.flatMap(line => line.split(",").zipWithIndex.map{ case ("true", i) => (i, "1") case ("false", i) => (i, "0") case p => p.swap }) val result = pairs.groupByKey.map{ case (k, vals) => { val valsString = vals.mkString(",") s"$k,$valsString" } } result.saveAsTextFile("scalaout")
Python:
def drop_first_line(index, itr): if index == 0: return iter(list(itr)[1:]) else: return itr def separate_cols(line): line = line.replace('true', '1').replace('false', '0') vals = line.split(',') for (i, x) in enumerate(vals): yield (i, x) input = (sc .textFile('train.csv', minPartitions=6) .mapPartitionsWithIndex(drop_first_line)) pairs = input.flatMap(separate_cols) result = (pairs .groupByKey() .map(lambda kv: "{0},{1}".format(kv[0], ",".join(kv[1])))) result.saveAsTextFile("pythonout")
Results
In local[6] mode (Intel(R) Xeon(R) CPU E3-1245 V2 @ 3.40GHz) with 4GB memory per executor it takes (n = 3):
- Scala - mean: 250.00s, stdev: 12.49
- Python - mean: 246.66s, stdev: 1.15
I am pretty sure that most of that time is spent on shuffling, serializing, deserializing and other secondary tasks. Just for fun, here’s naive single-threaded code in Python that performs the same task on this machine in less than a minute:
def go(): with open("train.csv") as fr: lines = [ line.replace('true', '1').replace('false', '0').split(",") for line in fr] return zip(*lines[1:])
