Skewed Window Function & Hive Source Partitions?

Question

The data I am reading via Spark is highly skewed Hive Table with the following stats.

(MIN, 25TH, MEDIAN, 75TH, MAX) via Spark UI:

1506.0 B / 0 232.4 KB / 27288 247.3 KB / 29025 371.0 KB / 42669 269.0 MB / 27197137

I believe it is causing problems downstream in the job when I perform some Window Funcs, and Pivots.

I tried exploring this parameter to limit the partition size however nothing changed and the partitions are still skewed upon read.

spark.conf.set("spark.sql.files.maxPartitionBytes")

Also, when I cache this DF with the Hive table as source it takes a few min and even causes some GC in the Spark UI most likely because of the skew as well.

Does this spark.sql.files.maxPartitionBytes work on Hive tables or only files?

What is the best course of action for handling this skewed Hive source?

Would something like a stage barrier write to parquet or Salting be suitable for this problem?

I would like to avoid .repartition() on read as it adds another layer to an already data roller-coaster of a job.

Thank you

==================================================

After further research it appears the Window Function is causing skewed data too and this is where the Spark Job hangs.

I am performing some time series filling via double Window Function (forward then backward fill to impute all the null sensor readings) and am trying to follow this article to try a salt method to evenly distribute ... however the following code produces all null values so the salt method is not working.

Not sure why I am getting skews after Window since each measure item I am partitioning by has roughly the same amount of records after checking via .groupBy() ... thus why would salt be needed?

+--------------------+-------+
|          measure   |  count|
+--------------------+-------+
|    v1              |5030265|
|      v2            |5009780|
|     v3             |5030526|
| v4                 |5030504|
...

salt post => https://medium.com/appsflyer/salting-your-spark-to-scale-e6f1c87dd18

nSaltBins = 300 # based off number of "measure" values
df_fill = df_fill.withColumn("salt", (F.rand() * nSaltBins).cast("int"))

# FILLS [FORWARD + BACKWARD]
window = Window.partitionBy('measure')\
               .orderBy('measure', 'date')\
               .rowsBetween(Window.unboundedPreceding, 0)

# FORWARD FILLING IMPUTER
ffill_imputer = F.last(df_fill['new_value'], ignorenulls=True)\
.over(window)
fill_measure_DF = df_fill.withColumn('value_impute_temp', ffill_imputer)\
.drop("value", "new_value")

window = Window.partitionBy('measure')\
               .orderBy('measure', 'date')\
               .rowsBetween(0,Window.unboundedFollowing)

# BACKWARD FILLING IMPUTER
bfill_imputer = F.first(df_fill['value_impute_temp'], ignorenulls=True)\
.over(window)
df_fill = df_fill.withColumn('value_impute_final', bfill_imputer)\
.drop("value_impute_temp")

user_647053 · Answer

Hive based solution :

You can enable Skew join optimization using hive configuration. Applicable settings are:

set hive.optimize.skewjoin=true;
set hive.skewjoin.key=500000;
set hive.skewjoin.mapjoin.map.tasks=10000;
set hive.skewjoin.mapjoin.min.split=33554432;

See databricks tips for this :

skew hints may work in this case

user_7351009 · Answer

Salting might be helpful in the case where a single partition is big enough to not fit in memory on a single executor. This might happen even if all the keys are equally distributed as well (as in your case).

You have to include the salt column in your partitionBy clause which you are using to create the Window.

window = Window.partitionBy('measure', 'salt')\
               .orderBy('measure', 'date')\
               .rowsBetween(Window.unboundedPreceding, 0)

Again you have to create another window which will operate on the intermediate result

window1 = Window.partitionBy('measure')\
                   .orderBy('measure', 'date')\
                   .rowsBetween(Window.unboundedPreceding, 0)

Skewed Window Function & Hive Source Partitions?

2 Answers