Why is my Spark DataFrame much slower than RDD?

Question

I have a very simple Spark DataFrame, and when running a DataFrame groupby, the performance is terrible - about 8x slower than the (in my head) equivalent RDD reduceByKey...

My cached DF is just two columns, customer and name with only 50k rows:

== Physical Plan ==
InMemoryColumnarTableScan [customer#2454,name#2456], InMemoryRelation [customer#2454,name#2456], true, 10000, StorageLevel(true, true, false, true, 1), Scan ParquetRelation[customer#2454,name#2456] InputPaths: hdfs://nameservice1/tmp/v2_selected_parquet/test_parquet2, None

When I run the following two snippets, I'd expect similar performance, not the rdd version to run in 10s and the DF version in 85s...

rawtempDF2.rdd.map(lambda x: (x['name'], 1)).reduceByKey(lambda x,y: x+y).collect()

rawtempDF2.groupby('name').count().collect()

Am I missing something really fundamental here? FWIW, the RDD version runs 54 stages, and the DF version is 227 :/

Edit: I'm using Spark 1.6.1 and Python 3.4.2. Edit2: Also, the source parquet was partitioned customer/day/name - currently 27 customers, 1 day, c. 45 names.

Answer 1

Both numbers seem to be relatively high and it is not exactly clear how you create DataFrame or measure the time but in general difference like this can be explained by a low number of records compared to the number of partitions.

Default value for spark.sql.shuffle.partitions is 200 which in the number of tasks you get. With 50K records an overhead of starting a task will be higher than a speedup you can gain from the parallel execution. Let's illustrate that with a simple example. First lets create an example data:

import string
import random

random.seed(323)

def random_string():
  n = random.randint(3, 6)
  return (''.join(random.choice(string.ascii_uppercase) for _ in range(n)), )

df = (sc
    .parallelize([random_string() for _ in range(50000)], 8).toDF(["name"])
    .cache())

And measure the time depending on number of shuffle.partitions:

sqlContext.setConf("spark.sql.shuffle.partitions", "1")
%timeit -n 10  df.groupby('name').count().collect()
## 10 loops, best of 3: 504 ms per loop

sqlContext.setConf("spark.sql.shuffle.partitions", "1")
%timeit -n 10  df.groupby('name').count().collect()
## 10 loops, best of 3: 451 ms per loop

sqlContext.setConf("spark.sql.shuffle.partitions", "100")
%timeit -n 10  df.groupby('name').count().collect()
## 10 loops, best of 3: 624 ms per loop

sqlContext.setConf("spark.sql.shuffle.partitions", "200")
%timeit -n 10  df.groupby('name').count().collect()
## 10 loops, best of 3: 778 ms per loop

sqlContext.setConf("spark.sql.shuffle.partitions", "1000")
%timeit -n 10  df.groupby('name').count().collect()
## 10 loops, best of 3: 1.75 s per loop

Although these values are not comparable to what you claim and this data has been collected in a local mode you can see that relatively clear pattern. The same applies to RDD:

from operator import add

%timeit -n 10 df.rdd.map(lambda x: (x['name'], 1)).reduceByKey(add, 1).collect()
## 10 loops, best of 3: 414 ms per loop

%timeit -n 10 df.rdd.map(lambda x: (x['name'], 1)).reduceByKey(add, 10).collect()
## 10 loops, best of 3: 439 ms per loop

%timeit -n 10 df.rdd.map(lambda x: (x['name'], 1)).reduceByKey(add, 100).collect()
## 10 loops, best of 3: 1.3 s per loop

%timeit -n 10 df.rdd.map(lambda x: (x['name'], 1)).reduceByKey(add, 1000).collect()
## 10 loops, best of 3: 8.41 s per loop

In a proper distributed environment this will be higher due to cost of the network IO.

Just for comparison lets check how long does it take to execute this task locally without Spark

from collections import Counter

data = df.rdd.flatMap(lambda x: x).collect()

%timeit -n 10 Counter(data)
## 10 loops, best of 3: 9.9 ms per loop

You should also take a look at the data locality. Depending on a storage you use and configuration this can add additional delay to your jobs even with a small input like this.