Spark SQL window function causes skew in data distribution

Question

The performance of this Spark SQL query is bad due to skew data distribution:

select c.*, coalesce(
      sum(revenue)
        OVER (PARTITION BY cid, pid, code
        ORDER BY (cTime div (1000*3600))
        RANGE BETWEEN 336 PRECEDING and 1 PRECEDING), 0L) as totalRevenue
  from records c

I see in SparkUI that single task stack and the cluster fail if I increase the scanned range.

I am using Yarn at AWS EMR, with Spark 2.2.0

How can I overcome this issue? Thanks

Answer 1

I can only recommend several approaches to alleviate your condition for investigation. I would actually try two approaches that don’t treat the skew first:

• Try increasing the executor memory per the message. On YARN you may additionally need to increase the maximum container memory as well. The default on Spark IIRC is 2gb and its not uncommon to need to increase it.
• Try switching to memory_and_disk or disk_only persistence levels. I believe this should work for your query although it can be hard to eyeball the full query plan

The reason for this is that at least to my eye your data is fundamentally skewed. You’re setting yourself up for maintenance difficulties if you start reshaping the data to address the skew in specific ways to the current shape of the data because the shape of the data may change over time. In my opinion at least you want to preserve the most straightforward implementation of your query for as long as you can, and only optimize skew issues programmatically if you hit problems with SLA violations, etc.

If those don’t work then you can try to address the skew directly. A simple approach for this is to create a third column that is populated by a random number for the column values that are known to be problematic. Do one pass of your summing operation with this in place, using it as a key, then a second pass with the extra random column removed. Alternatively you can do two queries and concatenate them: one with the random number for skewed data (which must still be handled in two passes) and another unaltered query for the non problematic data.

Edit - compute partial sums through two frames

The fundamentally useful observation here is that addition is commutative and associative. My original proposal based on random numbers won't work but this will. Basically, you want to compute the partial sum of the frame you want in several parts. The easiest way to do this is probably as a set of ranges (two used here for simplicity):

create temporary table partial_revenue_1 as select c.*, coalesce(
      sum(revenue)
        OVER (PARTITION BY cid, pid, code
        ORDER BY (cTime div (1000*3600))
        RANGE BETWEEN 336 PRECEDING and 118 PRECEDING), 0L) as partialTotalRevenue
  from records c

create temporary table partial_revenue_2 as select c.*, coalesce(
      sum(revenue)
        OVER (PARTITION BY cid, pid, code
        ORDER BY (cTime div (1000*3600))
        RANGE BETWEEN 117 PRECEDING and 1 PRECEDING), 0L) as partialTotalRevenue
  from records c

create temporary table combined_partials as select * from
    partial_reveneue_1 union all select * from partial_revenue_2

select sum(partialTotalRevenue), first(c.some_col) ... from 
    combined_partials c group by cid, pid, code

Notice you need to use the first aggregate function to cull the duplicate fields that you will have from the earlier select * operations on the records table. Don't worry, this will be fine since both values came from the same table.