Detect and deal with knowledge skew on AWS Glue

AWS Glue is a totally managed, serverless knowledge integration service supplied by Amazon Net Companies (AWS) that makes use of Apache Spark as one among its backend processing engines (as of this writing, you should use Python Shell, Spark, or Ray).

Knowledge skew happens when the information being processed will not be evenly distributed throughout the Spark cluster, inflicting some duties to take considerably longer to finish than others. This will result in inefficient useful resource utilization, longer processing instances, and finally, slower efficiency. Knowledge skew can come up from varied components, together with uneven knowledge distribution, skewed be a part of keys, or uneven knowledge processing patterns. Despite the fact that the most important challenge is commonly having nodes working out of disk throughout shuffling, which ends up in nodes falling like dominoes and job failures, it’s additionally necessary to say that knowledge skew is hidden. The stealthy nature of knowledge skew means it might probably typically go undetected as a result of monitoring instruments may not flag an uneven distribution as a vital challenge, and logs don’t all the time make it evident. Consequently, a developer could observe that their AWS Glue jobs are finishing with out obvious errors, but the system might be working removed from its optimum effectivity. This hidden inefficiency not solely will increase operational prices resulting from longer runtimes however also can result in unpredictable efficiency points which can be tough to diagnose with out a deep dive into the information distribution and process run patterns.

For instance, in a dataset of buyer transactions, if one buyer has considerably extra transactions than the others, it might probably trigger a skew within the knowledge distribution.

Figuring out and dealing with knowledge skew points is essential to having good efficiency on Apache Spark and due to this fact on AWS Glue jobs that use Spark as a backend. On this put up, we present how one can determine knowledge skew and focus on the completely different methods to mitigate knowledge skew.

How one can detect knowledge skew

When an AWS Glue job has points with native disks (cut up disk points), doesn’t scale with the variety of staff, or has low CPU utilization (you possibly can allow Amazon CloudWatch metrics on your job to have the ability to see this), you could have an information skew challenge. You possibly can detect knowledge skew with knowledge evaluation or by utilizing the Spark UI. On this part, we focus on learn how to use the Spark UI.

The Spark UI offers a complete view of Spark functions, together with the variety of duties, phases, and their period. To make use of it you must allow Spark UI occasion logs on your job runs. It’s enabled by default on Glue console and as soon as enabled, Spark occasion log information can be created throughout the job run and saved in your S3 bucket. Then, these logs are parsed, and you should use the AWS Glue serverless Spark UI to visualise them. You possibly can consult with this blogpost for extra particulars. In these jobs the place the AWS Glue serverless Spark UI doesn’t work because it has a restrict of 512 MB of logs, you possibly can arrange the Spark UI utilizing an EC2 occasion.

You need to use the Spark UI to determine which duties are taking longer to finish than others, and if the information distribution amongst partitions is balanced or not (do not forget that in Spark, one partition is mapped to 1 process). If there may be knowledge skew, you will notice that some partitions have considerably extra knowledge than others. The next determine reveals an instance of this. We are able to see that one process is taking much more time than the others, which might point out knowledge skew.

One other factor that you should use is the abstract metrics for every stage. The next screenshot reveals one other instance of knowledge skew.

These metrics signify the task-related metrics beneath which a sure share of duties accomplished. For instance, the seventy fifth percentile process period signifies that 75% of duties accomplished in much less time than this worth. When the duties are evenly distributed, you will notice comparable numbers in all of the percentiles. When there may be knowledge skew, you will notice very biased values in every percentile. Within the previous instance, it didn’t write many shuffle information (lower than 50 MiB) in Min, twenty fifth percentile, Median, and seventy fifth percentile. Nevertheless, in Max, it wrote 460 MiB, 10 instances the seventy fifth percentile. It means there was a minimum of one process (or as much as 25% of duties) that wrote a lot greater shuffle information than the remainder of the duties. It’s also possible to see that the period of the tax in Max is 46 seconds and the Median is 2 seconds. These are all indicators that your dataset could have knowledge skew.

AWS Glue interactive periods

You need to use interactive periods to load your knowledge from the AWS Glue Knowledge Catalog or simply use Spark strategies to load the information similar to Parquet or CSV that you simply wish to analyze. You need to use an identical script to the next to detect knowledge skew from the partition measurement perspective; the extra necessary challenge is said to knowledge skew whereas shuffling, and this script doesn’t detect that type of skew:

from pyspark.sql.features import spark_partition_id, asc, desc
#input_dataframe being the dataframe the place you wish to examine for knowledge skew
partition_sizes_df=input_dataframe
    .withColumn("partitionId", spark_partition_id())
    .groupBy("partitionId")
    .depend()
    .orderBy(asc("depend"))
    .withColumnRenamed("depend","partition_size")
#calculate common and standar deviation for the partition sizes
avg_size = partition_sizes_df.agg({"partition_size": "avg"}).acquire()[0][0]
std_dev_size = partition_sizes_df.agg({"partition_size": "stddev"}).acquire()[0][0]

""" 
 the code calculates absolutely the distinction between every worth within the "partition_size" column and the calculated common (avg_size).
 then, calculates twice the usual deviation (std_dev_size) and use 
 that as a boolean masks the place the situation checks if absolutely the distinction is larger than twice the usual deviation
 with a view to mark a partition 'skewed'
"""
skewed_partitions_df = partition_sizes_df.filter(abs(partition_sizes_df["partition_size"] - avg_size) > 2 * std_dev_size)
if skewed_partitions_df.depend() > 0:
    skewed_partitions = [row["partition_id"] for row in skewed_partitions_df.acquire()]
    print(f"The next partitions have considerably completely different sizes: {skewed_partitions}")
else:
    print("No knowledge skew detected.")

You possibly can calculate the common and normal deviation of partition sizes utilizing the agg() perform and determine partitions with considerably completely different sizes utilizing the filter() perform, and you’ll print their indexes if any skewed partitions are detected. In any other case, the output prints that no knowledge skew is detected.

This code assumes that your knowledge is structured, and chances are you’ll want to change it in case your knowledge is of a special sort.

How one can deal with knowledge skew

You need to use completely different methods in AWS Glue to deal with knowledge skew; there is no such thing as a single common answer. The very first thing to do is verify that you simply’re utilizing newest AWS Glue model, for instance AWS Glue 4.0 based mostly on Spark 3.3 has enabled by default some configs like Adaptative Question Execution (AQE) that may assist enhance efficiency when knowledge skew is current.

The next are among the methods that you may make use of to deal with knowledge skew:

• Filter and carry out – If you realize which keys are inflicting the skew, you possibly can filter them out, carry out your operations on the non-skewed knowledge, after which deal with the skewed keys individually.
• Implementing incremental aggregation – In case you are performing a big aggregation operation, you possibly can break it up into smaller phases as a result of in giant datasets, a single aggregation operation (like sum, common, or depend) might be resource-intensive. In these circumstances, you possibly can carry out intermediate actions. This might contain filtering, grouping, or extra aggregations. This can assist distribute the workload throughout the nodes and cut back the dimensions of intermediate knowledge.
• Utilizing a customized partitioner – In case your knowledge has a selected construction or distribution, you possibly can create a customized partitioner that partitions your knowledge based mostly on its traits. This can assist guarantee that knowledge with comparable traits is in the identical partition and cut back the dimensions of the most important partition.
• Utilizing broadcast be a part of – In case your dataset is small however exceeds the spark.sql.autoBroadcastJoinThreshold worth (default is 10 MB), you may have the choice to both present a touch to make use of broadcast be a part of or regulate the brink worth to accommodate your dataset. This may be an efficient technique to optimize be a part of operations and mitigate knowledge skew points ensuing from shuffling giant quantities of knowledge throughout nodes.
• Salting – This includes including a random prefix to the important thing of skewed knowledge. By doing this, you distribute the information extra evenly throughout the partitions. After processing, you possibly can take away the prefix to get the unique key values.

These are just some methods to deal with knowledge skew in PySpark; one of the best method will rely upon the traits of your knowledge and the operations you’re performing.

The next is an instance of becoming a member of skewed knowledge with the salting approach:

from pyspark.sql import SparkSession
from pyspark.sql.features import lit, ceil, rand, concat, col

# Outline the variety of salt values
num_salts = 3

# Perform to determine skewed keys
def identify_skewed_keys(df, key_column, threshold):
    key_counts = df.groupBy(key_column).depend()
    return key_counts.filter(key_counts['count'] > threshold).choose(key_column)

# Determine skewed keys
skewed_keys = identify_skewed_keys(skewed_data, "key", skew_threshold)

# Splitting the dataset
skewed_data_subset = skewed_data.be a part of(skewed_keys, ["key"], "interior")
non_skewed_data_subset = skewed_data.be a part of(skewed_keys, ["key"], "left_anti")

# Apply salting to skewed knowledge
skewed_data_subset = skewed_data_subset.withColumn("salt", ceil((rand() * 10) % num_salts))
skewed_data_subset = skewed_data_subset.withColumn("salted_key", concat(col("key"), lit("_"), col("salt")))

# Replicate skewed rows in non-skewed dataset
def replicate_skewed_rows(df, keys, multiplier):
    replicated_df = df.be a part of(keys, ["key"]).crossJoin(spark.vary(multiplier).withColumnRenamed("id", "salt"))
    replicated_df = replicated_df.withColumn("salted_key", concat(col("key"), lit("_"), col("salt")))
    return replicated_df.drop("salt")

replicated_non_skewed_data = replicate_skewed_rows(non_skewed_data, skewed_keys, num_salts)

# Carry out the JOIN operation on the salted keys for skewed knowledge
result_skewed = skewed_data_subset.be a part of(replicated_non_skewed_data, "salted_key")

# Carry out common be a part of on non-skewed knowledge
result_non_skewed = non_skewed_data_subset.be a part of(non_skewed_data, "key")

# Mix outcomes
final_result = result_skewed.union(result_non_skewed)

On this code, we first outline a salt worth, which is usually a random integer or another worth. We then add a salt column to our DataFrame utilizing the withColumn() perform, the place we set the worth of the salt column to a random quantity utilizing the rand() perform with a hard and fast seed. The perform replicate_salt_rows is outlined to duplicate every row within the non-skewed dataset (non_skewed_data) num_salts instances. This ensures that every key within the non-skewed knowledge has matching salted keys. Lastly, a be a part of operation is carried out on the salted_key column between the skewed and non-skewed datasets. This be a part of is extra balanced in comparison with a direct be a part of on the unique key, as a result of salting and replication have mitigated the information skew.

The rand() perform used on this instance generates a random quantity between 0–1 for every row, so it’s necessary to make use of a hard and fast seed to attain constant outcomes throughout completely different runs of the code. You possibly can select any mounted integer worth for the seed.

The next figures illustrate the information distribution earlier than (left) and after (proper) salting. Closely skewed key2 recognized and salted into key2_0, key2_1, and key2_2, balancing the information distribution and stopping any single node from being overloaded. After processing, the outcomes might be aggregated again, in order that that the ultimate output is in line with the unsalted key values.

Different methods to make use of on skewed knowledge throughout the be a part of operation

Once you’re performing skewed joins, you should use salting or broadcasting methods, or divide your knowledge into skewed and common components earlier than becoming a member of the common knowledge and broadcasting the skewed knowledge.

In case you are utilizing Spark 3, there are computerized optimizations for attempting to optimize Knowledge Skew points on joins. These might be tuned as a result of they’ve devoted configs on Apache Spark.

Conclusion

This put up supplied particulars on learn how to detect knowledge skew in your knowledge integration jobs utilizing AWS Glue and completely different methods for dealing with it. Having a superb knowledge distribution is essential to reaching one of the best efficiency on distributed processing programs like Apache Spark.

Though this put up targeted on AWS Glue, the identical ideas apply to jobs chances are you’ll be working on Amazon EMR utilizing Apache Spark or Amazon Athena for Apache Spark.

As all the time, AWS welcomes your suggestions. Please go away your feedback and questions within the feedback part.

In regards to the Authors

Salim Tutuncu is a Sr. PSA Specialist on Knowledge & AI, based mostly from Amsterdam with a give attention to the EMEA North and EMEA Central areas. With a wealthy background within the know-how sector that spans roles as a Knowledge Engineer, Knowledge Scientist, and Machine Studying Engineer, Salim has constructed a formidable experience in navigating the complicated panorama of knowledge and synthetic intelligence. His present function includes working intently with companions to develop long-term, worthwhile companies leveraging the AWS Platform, significantly in Knowledge and AI use circumstances.

Angel Conde Manjon is a Sr. PSA Specialist on Knowledge & AI, based mostly in Madrid, and focuses on EMEA South and Israel. He has beforehand labored on analysis associated to Knowledge Analytics and Synthetic Intelligence in various European analysis tasks. In his present function, Angel helps companions develop companies centered on Knowledge and AI.