Python vs. Scala for Apache Spark - expected benchmark with unexpected result
Apache Spark today is perhaps the most popular platform for analyzing large-volume data. A considerable contribution to its popularity is the possibility of using from under Python. At the same time, everyone agrees that, within the framework of the standard API, the performance of Python and Scala / Java code is comparable, but there is no single point of view regarding user-defined functions (User Defined Function, UDF). Let's try to figure out how the overhead in this case increases, using the example of the task of checking the SNA Hackathon 2019 solution .
As part of the competition, participants solve the problem of sorting the news feed of a social network and upload solutions in the form of a set of sorted lists. To check the quality of the solution obtained, first, for each of the loaded lists, the ROC AUC is calculated , and then the average value is displayed. Please note that you need to calculate not one common ROC AUC, but a personal one for each user - there is no ready-made design for solving this problem, so you will have to write a specialized function. A good reason to compare the two approaches in practice.
As a comparison platform, we will use a cloud container with four cores and Spark launched in local mode, and we will work with it through Apache Zeppelin . To compare the functionality, we will mirror the same code in PySpark and Scala Spark. [here] Let's start by loading the data.
data = sqlContext.read.csv("sna2019/modelCappedSubmit")
trueData = sqlContext.read.csv("sna2019/collabGt")
toValidate = data.withColumnRenamed("_c1", "submit") \
.join(trueData.withColumnRenamed("_c1", "real"), "_c0") \
.withColumnRenamed("_c0", "user") \
.repartition(4).cache()
toValidate.count()val data = sqlContext.read.csv("sna2019/modelCappedSubmit")
val trueData = sqlContext.read.csv("sna2019/collabGt")
val toValidate = data.withColumnRenamed("_c1", "submit")
.join(trueData.withColumnRenamed("_c1", "real"), "_c0")
.withColumnRenamed("_c0", "user")
.repartition(4).cache()
toValidate.count()When using the standard API, the almost complete identity of the code, up to the keyword, attracts attention val. Operating time is not significantly different. Now let's try to determine the UDF we need.
parse = sqlContext.udf.register("parse",
lambda x: [int(s.strip()) for s in x[1:-1].split(",")], ArrayType(IntegerType()))
def auc(submit, real):
trueSet = set(real)
scores = [1.0 / (i + 1) for i,x in enumerate(submit)]
labels = [1.0 if x in trueSet else 0.0 for x in submit]
return float(roc_auc_score(labels, scores))
auc_udf = sqlContext.udf.register("auc", auc, DoubleType())val parse = sqlContext.udf.register("parse",
(x : String) => x.slice(1,x.size - 1).split(",").map(_.trim.toInt))
case class AucAccumulator(height: Int, area: Int, negatives: Int)
val auc_udf = sqlContext.udf.register("auc", (byScore: Seq[Int], gt: Seq[Int]) => {
val byLabel = gt.toSet
val accumulator = byScore.foldLeft(AucAccumulator(0, 0, 0))((accumulated, current) => {
if (byLabel.contains(current)) {
accumulated.copy(height = accumulated.height + 1)
} else {
accumulated.copy(area = accumulated.area + accumulated.height, negatives = accumulated.negatives + 1)
}
})
(accumulator.area).toDouble / (accumulator.negatives * accumulator.height)
})When implementing a specific function, it is clear that Python is more concise, primarily because of the ability to use the built-in scikit-learn function . However, there are unpleasant moments - you must explicitly specify the type of the return value, whereas in Scala it is automatically determined. Let's perform the operation:
toValidate.select(auc_udf(parse("submit"), parse("real"))).groupBy().avg().show()toValidate.select(auc_udf(parse($"submit"), parse($"real"))).groupBy().avg().show()The code looks almost identical, but the results are discouraging.
The implementation on PySpark worked out one and a half minutes instead of two seconds on Scala, that is, Python turned out to be 45 times slower . While running, top shows 4 active Python processes that are running at full speed, and this suggests that the Global Interpreter Lock does not create problems here . But! Perhaps the problem is in the internal implementation of scikit-learn - let's try to reproduce the Python code literally, without resorting to the standard libraries.
def auc(submit, real):
trueSet = set(real)
height = 0
area = 0
negatives = 0
for candidate in submit:
if candidate in trueSet:
height = height + 1
else:
area = area + height
negatives = negatives + 1
return float(area) / (negatives * height)
auc_udf_modified = sqlContext.udf.register("auc_modified", auc, DoubleType())
toValidate.select(auc_udf_modified(parse("submit"), parse("real"))).groupBy().avg().show()
The experiment shows interesting results. On the one hand, with this approach, productivity was leveled, but on the other, laconicism disappeared. The results obtained may indicate that when working in Python using additional C ++ modules, significant overheads appear for switching between contexts. Of course, there is similar overhead when using JNI in Java / Scala, however, I did not have to deal with examples of degradation 45 times when using them.
For a more detailed analysis, we will carry out two additional experiments: using pure Python without Spark to measure the contribution from a package call, and with an increased data size in Spark to amortize overhead and get a more accurate comparison.
def parse(x):
return [int(s.strip()) for s in x[1:-1].split(",")]
def auc(submit, real):
trueSet = set(real)
height = 0
area = 0
negatives = 0
for candidate in submit:
if candidate in trueSet:
height = height + 1
else:
area = area + height
negatives = negatives + 1
return float(area) / (negatives * height)
def sklearn_auc(submit, real):
trueSet = set(real)
scores = [1.0 / (i + 1) for i,x in enumerate(submit)]
labels = [1.0 if x in trueSet else 0.0 for x in submit]
return float(roc_auc_score(labels, scores))
The experiment with local Python and Pandas confirmed the assumption of significant overhead when using additional packages - when using scikit-learn, the speed decreases by more than 20 times. However, 20 is not 45 - let's try to “inflate” the data and compare Spark performance again.
k4 = toValidate.union(toValidate)
k8 = k4.union(k4)
m1 = k8.union(k8)
m2 = m1.union(m1)
m4 = m2.union(m2).repartition(4).cache()
m4.count()
The new comparison shows the speed advantage of the Scala implementation over Python by 7-8 times - 7 seconds versus 55. Finally, let's try "the fastest that is in Python" - numpy to calculate the sum of the array:
import numpy
numpy_sum = sqlContext.udf.register("numpy_sum",
lambda x: float(numpy.sum(x)), DoubleType())val my_sum = sqlContext.udf.register("my_sum", (x: Seq[Int]) => x.map(_.toDouble).sum)
Again a significant slowdown - 5 seconds of Scala versus 80 seconds of Python. Summing up, we can draw the following conclusions:
- While PySpark operates within the framework of the standard API, in terms of speed it can really be compared to Scala.
- When specific logic appears in the form of User Defined Functions, PySpark performance noticeably decreases. With enough information, when the processing time for a data block exceeds several seconds, the Python implementation is 5-10 slower due to the need to move data between processes and spend resources on interpreting Python.
- If the use of additional functions implemented in C ++ modules appears, then additional call costs arise, and the difference between Python and Scala increases up to 10-50 times.
As a result, despite all the charm of Python, its use in conjunction with Spark does not always look justified. If there is not much data to make Python overhead significant, then you should think about whether Spark is needed here? If there is a lot of data, but processing occurs within the framework of the standard Spark SQL API, then is Python needed here?
If there is a lot of data and often have to deal with tasks that go beyond the limits of the SQL API, then in order to perform the same amount of work when using PySpark, you will have to increase the cluster at times. For example, for Odnoklassniki, the cost of capital expenditures for the Spark cluster would increase by many hundreds of millions of rubles. And if you try to take advantage of the advanced capabilities of the Python ecosystem libraries, that is, the risk of slowing down is not just at times, but an order of magnitude.
Some acceleration can be obtained using the relatively new functionality of vectorized functions. In this case, not a single row is fed to the UDF input, but a packet of several rows in the form of a Pandas Dataframe. However, the development of this functionality is not yet completed , and even in this case the difference will be significant .
An alternative would be to maintain an extensive team of data engineers who can quickly address the needs of data scientists with additional features. Or to immerse yourself in the Scala world, since it’s not so difficult: many of the necessary tools already exist , training programs appear that go beyond PySpark.
