Spark dataframe distinct write is increasing the output size by almost 10 fold

user_5199321 0
Viewed 96

I have a case where i am trying to write some results using dataframe write into S3 using the below query with input_table_1 size is 13 Gb and input_table_2 as 1 Mb

input_table_1 has columns account, membership and input_table_2 has columns role, id , membership_id, quantity, start_date

SELECT 
    /*+ BROADCASTJOIN(input_table_2) */
    account,
    role,
    id,
    quantity,
    cast(start_date AS string) AS start_date
FROM
    input_table_1
    INNER JOIN
        input_table_2
        ON array_contains(input_table_1.membership, input_table_2.membership_id)

where membership array contains list of member_ids

This dataset write using Spark dataframe is generating around 1.1TiB of data in S3 with around 700 billion records.

We identified that there are duplicates and used dataframe.distinct.write.parquet("s3path") to remove the duplicates . The record count is reduced to almost 1/3rd of the previous total count with around 200 billion rows but we observed that the output size in S3 is now 17.2 TiB .

I am very confused how this can happen.

I have used the following spark conf settings

spark.sql.shuffle.partitions=20000

I have tried to do a coalesce and write to s3 but it did not work.

Please suggest if this is expected and when can be done ?

apache-spark-sql apache-spark
1 Answers

There's two sides to this:

1) Physical translation of distinct in Spark

The Spark catalyst optimiser turns a distinct operation into an aggregation by means of the ReplaceDeduplicateWithAggregate rule (Note: in the execution plan distinct is named Deduplicate).

This basically means df.distinct() on all columns is translated into a groupBy on all columns with an empty aggregation: df.groupBy(df.columns:_*).agg(Map.empty).

Spark uses a HashPartitioner when shuffling data for a groupBy on respective columns. Since the groupBy clause in your case contains all columns (well, implicitly, but it does), you're more or less randomly shuffling data to different nodes in the cluster.

Increasing spark.sql.shuffle.partitions in this case is not going to help.

Now on to the 2nd side, why does this affect the size of your parquet files so much?

2) Compression in parquet files

Parquet is a columnar format, will say your data is organised in columns rather than row by row. This allows for powerful compression if data is adequately laid-out & ordered. E.g. if a column contains the same value for a number of consecutive rows, it is enough to write that value just once and make a note of the number of repetitions (a strategy called run length encoding). But Parquet also uses various other compression strategies.

Unfortunately, data ends up pretty randomly in your case after shuffling to remove duplicates. The original partitioning of input_table_1 was much better fitted.

Solutions

There's no single answer how to solve this, but here's a few pointers I'd suggest doing next:

  • What's causing the duplicates? Could these be removed upstream? Or is there a problem with the join condition causing duplicates?

  • A simple solution is to just repartition the dataset after distinct to match the partitioning of your input data. Adding a secondary sorting (sortWithinPartition) is likely going to give you even better compression. However, this comes at the cost of an additional shuffle!

    • As @matt-andruff pointed out below, you can also achieve this in SQL using cluster by. Obviously, that also requires you to move the distinct keyword into your SQL statement.

  • Write your own deduplication algorithm as Spark Aggregator and group / shuffle the data just once in a meaningful way.

user_42780320
Related