topic Re: Skewness / Salting with countDistinct in Data Engineering

Skewness / Salting with countDistinct

KosmaS — Mon, 05 Aug 2024 15:25:15 GMT

Hey Everyone,

I experience data skewness for:

df = (source_df .unionByName(source_df.withColumn("region", lit("Country"))) .groupBy("zip_code", "region", "device_type") .agg(countDistinct("device_id").alias("total_active_unique"), count("device_id").alias("total_active"))

Stats:

Is there a way to work with data skewness where I need to calculate a countDistinct in aggregation and avoid affecting the results?

I understand how to work with data skewness by adding salting, but it seems to be fine with count.
But when countDistinct comes to the picture salting seems to be affecting the results.
Is there some tricky way to still apply salting and secure deterministic results for countDistinct?
Or is there some other approach in such case to be applied for data skewness?

Re: Skewness / Salting with countDistinct

Retired_mod — Wed, 07 Aug 2024 13:30:55 GMT

Hi @KosmaS, To address data skewness with `countDistinct`, you can use several techniques:

Double Aggregation involves salting the data, performing an aggregation, then removing the salt and aggregating again to reduce skewness.

HyperLogLog (HLL) provides approximate results for `countDistinct`, balancing accuracy and performance.

Broadcast Joins can help with small skewed datasets by reducing shuffle issues, while Partitioning the data based on skewed keys can distribute the load more evenly.

If you have any more details or specific constraints, feel free to share!

Re: Skewness / Salting with countDistinct

KosmaS — Thu, 22 Aug 2024 08:17:08 GMT

Hey @Retired_mod

thanks for the reply. I tried to spend some time on your response.

You're suggesting 'double aggregation' and as I'd be guessing it should look more or less this way:

df = (source_df .unionByName(source_df.withColumn("region", lit("Country"))) .groupBy("zip_code", "region", "device_type", "salt") .agg(countDistinct("device_id").alias("total_active_unique"), count("device_id").alias("total_active")) .groupBy("zip_code", "region", "device_type") .agg(countDistinct("device_id").alias("total_active_unique"), count("device_id").alias("total_active"))

So I can't see how countDistinct value won't be affected by salt. It'll be affected at the first step (with salt), so the second step will have inaccurate results. Or should this be done a bit differently? And did you mean something else?

Re: Skewness / Salting with countDistinct

singhvikash86 — Sat, 18 Jan 2025 19:24:29 GMT

What about salt function is function on device_id produces mutually exclusive results like hash(device_id) % 101 and then one more aggregation to sum of these counts group by zip_code, region, device_type

Re: Skewness / Salting with countDistinct

Avinash_Narala — Mon, 20 Jan 2025 11:52:12 GMT

you can make use of databricks native feature "Liquid Clustering", cluster by the columns which you want to use in grouping statements, it will handle the performance issue due to data skewness .

For more information, please do visit :

https://docs.databricks.com/en/delta/clustering.html