32

I am joining two big datasets using Spark RDD. One dataset is very much skewed so few of the executor tasks taking a long time to finish the job. How can I solve this scenario?

6 Answers 6

28

Pretty good article on how it can be done: https://datarus.wordpress.com/2015/05/04/fighting-the-skew-in-spark/

Short version:

  • Add random element to large RDD and create new join key with it
  • Add random element to small RDD using explode/flatMap to increase number of entries and create new join key
  • Join RDDs on new join key which will now be distributed better due to random seeding
5
  • Went through the article you mentioned. In my case, larger table has 23500 partitions with 2 billion records. Smaller table has 5 million records. How I decide value of 'N'? Any suggestions?
    – kalpesh
    May 7, 2018 at 10:25
  • 2
    @kalpesh, there is one more parameter that should be factored in, the size of a shuffle block in Spark should be less than 2GB SPARK-6235. I would recommend to focus on the size of partition which typically should be ~128MB instead of number of partitions. I have seen applications working fine with many partitions (> 32k).
    – Kyr
    May 31, 2018 at 12:25
  • @kalpesh Well you want to create enough partitions on the new keys that all your executors will get data to join to maximize parallelism.
    – LiMuBei
    Jun 4, 2018 at 12:58
  • 1
    @LiMuBei there is something I don't get in the python code small_rdd_transformed = small_rdd.cartesian(sc.parallelize(range(0, N))).map(lambda x: ((x[0][0], x[1]), x[0][1])).coalesce(num_parts).cache() # replicate the small rdd in scala x is a tuple2 so what does x[0][0] mean for instance?..
    – Bill'o
    Sep 6, 2018 at 18:20
  • my mistake there is a mapPartitionWithIndex missing
    – Bill'o
    Sep 6, 2018 at 19:11
19

Say you have to join two tables A and B on A.id=B.id. Lets assume that table A has skew on id=1.

i.e. select A.id from A join B on A.id = B.id

There are two basic approaches to solve the skew join issue:

Approach 1:

Break your query/dataset into 2 parts - one containing only skew and the other containing non skewed data. In the above example. query will become -

 1. select A.id from A join B on A.id = B.id where A.id <> 1;
 2. select A.id from A join B on A.id = B.id where A.id = 1 and B.id = 1;

The first query will not have any skew, so all the tasks of ResultStage will finish at roughly the same time.

If we assume that B has only few rows with B.id = 1, then it will fit into memory. So Second query will be converted to a broadcast join. This is also called Map-side join in Hive.

Reference: https://cwiki.apache.org/confluence/display/Hive/Skewed+Join+Optimization

The partial results of the two queries can then be merged to get the final results.

Approach 2:

Also mentioned by LeMuBei above, the 2nd approach tries to randomize the join key by appending extra column. Steps:

  1. Add a column in the larger table (A), say skewLeft and populate it with random numbers between 0 to N-1 for all the rows.

  2. Add a column in the smaller table (B), say skewRight. Replicate the smaller table N times. So values in new skewRight column will vary from 0 to N-1 for each copy of original data. For this, you can use the explode sql/dataset operator.

After 1 and 2, join the 2 datasets/tables with join condition updated to-

                *A.id = B.id && A.skewLeft = B.skewRight*

Reference: https://datarus.wordpress.com/2015/05/04/fighting-the-skew-in-spark/

3
16

Depending on the particular kind of skew you're experiencing, there may be different ways to solve it. The basic idea is:

  • Modify your join column, or create a new join column, that is not skewed but which still retains adequate information to do the join
  • Do the join on that non-skewed column -- resulting partitions will not be skewed
  • Following the join, you can update the join column back to your preferred format, or drop it if you created a new column

The "Fighting the Skew In Spark" article referenced in LiMuBei's answer is a good technique if the skewed data participates in the join. In my case, skew was caused by a very large number of null values in the join column. The null values were not participating in the join, but since Spark partitions on the join column, the post-join partitions were very skewed as there was one gigantic partition containing all of the nulls.

I solved it by adding a new column which changed all null values to a well-distributed temporary value, such as "NULL_VALUE_X", where X is replaced by random numbers between say 1 and 10,000, e.g. (in Java):

// Before the join, create a join column with well-distributed temporary values for null swids.  This column
// will be dropped after the join.  We need to do this so the post-join partitions will be well-distributed,
// and not have a giant partition with all null swids.
String swidWithDistributedNulls = "swid_with_distributed_nulls";
int numNullValues = 10000; // Just use a number that will always be bigger than number of partitions
Column swidWithDistributedNullsCol =
    when(csDataset.col(CS_COL_SWID).isNull(), functions.concat(
        functions.lit("NULL_SWID_"),
        functions.round(functions.rand().multiply(numNullValues)))
    )
    .otherwise(csDataset.col(CS_COL_SWID));
csDataset = csDataset.withColumn(swidWithDistributedNulls, swidWithDistributedNullsCol);

Then joining on this new column, and then after the join:

outputDataset.drop(swidWithDistributedNullsCol);
4

Taking reference from https://datarus.wordpress.com/2015/05/04/fighting-the-skew-in-spark/ below is the code for fighting the skew in spark using Pyspark dataframe API

Creating the 2 dataframes:

from math import exp
from random import randint
from datetime import datetime

def count_elements(splitIndex, iterator):
    n = sum(1 for _ in iterator)
    yield (splitIndex, n)

def get_part_index(splitIndex, iterator):
    for it in iterator:
        yield (splitIndex, it)

num_parts = 18
# create the large skewed rdd
skewed_large_rdd = sc.parallelize(range(0,num_parts), num_parts).flatMap(lambda x: range(0, int(exp(x))))
skewed_large_rdd = skewed_large_rdd.mapPartitionsWithIndex(lambda ind, x: get_part_index(ind, x))

skewed_large_df = spark.createDataFrame(skewed_large_rdd,['x','y'])

small_rdd = sc.parallelize(range(0,num_parts), num_parts).map(lambda x: (x, x))

small_df = spark.createDataFrame(small_rdd,['a','b'])

Dividing the data into 100 bins for large df and replicating the small df 100 times

salt_bins = 100
from pyspark.sql import functions as F

skewed_transformed_df = skewed_large_df.withColumn('salt', (F.rand()*salt_bins).cast('int')).cache()

small_transformed_df = small_df.withColumn('replicate', F.array([F.lit(i) for i in range(salt_bins)]))

small_transformed_df = small_transformed_df.select('*', F.explode('replicate').alias('salt')).drop('replicate').cache()

Finally the join avoiding the skew

t0 = datetime.now()
result2 = skewed_transformed_df.join(small_transformed_df, (skewed_transformed_df['x'] == small_transformed_df['a']) & (skewed_transformed_df['salt'] == small_transformed_df['salt']) )
result2.count() 
print "The direct join takes %s"%(str(datetime.now() - t0))
4

Apache DataFu has two methods for doing skewed joins that implement some of the suggestions in the previous answers.

The joinSkewed method does salting (adding a random number column to split the skewed values).

The broadcastJoinSkewed method is for when you can divide the dataframe into skewed and regular parts, as described in Approach 2 from the answer by moriarty007.

These methods in DataFu are useful for projects using Spark 2.x. If you are already on Spark 3, there are dedicated methods for doing skewed joins.

Full disclosure - I am a member of Apache DataFu.

0

You could try to repartition the "skewed" RDD to more partitions, or try to increase spark.sql.shuffle.partitions (which is by default 200).

In your case, I would try to set the number of partitions to be much higher than the number of executors.

4
  • 4
    spark.sql.shuffle.partitions will not help skewed data. There will be 200 partitions but only a few of them will have data.
    – Luniam
    Feb 22, 2018 at 2:15
  • 1
    Increasing spark.sql.shuffle.partitions to a higher number won't help with Skew. All the data corresponding to skewed key will still go to same reducer and will cause the slowness. Sep 13, 2018 at 8:07
  • 2
    @prakharjain this is not entirely true. increasing number partitions will reduce the chance that two keys with many record will be put in the same partition. Sep 13, 2018 at 8:35
  • 1
    spark.sql.shuffle.partitions will not do the trick. join on skewed data will cause hot spotting issue on data shuffling because the same value on the join point will be hashed into the same hash-key and due to that, all those rows(with the same value) will be sent to the same executor during the shuffling. refer: oreilly.com/library/view/high-performance-spark/9781491943199/…
    – linehrr
    Sep 21, 2018 at 19:15

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.