Overcoming Caching Challenges in Flattening Data Across Multiple Tables with AWS Glue

Introduction:

Working with large-scale data in distributed environments like AWS Glue is a complex task that often involves integrating, transforming, and flattening data from multiple tables. In one of my recent projects, I faced several challenges while flattening data from a parent table with multiple child tables. Along the way, I learned valuable lessons about memory management, Spark configurations, and optimizing performance. This post outlines the journey, challenges, and key learnings that could help others facing similar issues.

The Project:

The task was to flatten data from a parent table that had 28 child tables. This required performing a series of left joins on the child tables with the parent table, followed by concatenating multiple values in columns using a separator when multiple records were created for a single primary key. The goal was to produce a final flattened table with the same number of rows as the parent table but with all relevant data from the child tables included.

The Initial Approach:

Initially, I attempted to cache the flattened DataFrame to optimize performance. Caching worked fine when I used AWS Glue with G.8X DPUs, specifically a configuration with 10 G.8X DPUs, resulting in 80 total DPUs. In this configuration, each DPU had ample memory (128 GB per DPU), which allowed the caching operation to complete successfully.

Why Caching Worked:

In the G.8X configuration:

• High Memory per Executor: Each executor had significant memory (128 GB), allowing large DataFrames to be cached in memory without spilling to disk.
• Efficient Resource Utilization: With fewer executors handling larger chunks of data, memory-intensive operations like caching were managed efficiently.

The Problem:

A few months after the initial successful run as a POC, my lead asked me to switch to a different DPU for a different job configuration using G.4X DPUs I used the same configuration for this job too. Since it had been some time since my initial test, I didn't pay much attention to the instance type and focused instead on ensuring that the total DPU count remained the same—80 DPUs. I assumed that since both configurations resulted in the same number of total DPUs, the performance would be similar.

However, despite the same total DPU count, caching started to fail with out-of-memory errors. This was unexpected since the total memory across the cluster was supposed to be the same.

Realizing this was an issue with how the resources were allocated, I attempted to increase the number of DPUs to 120 and even 160 by using more G.4X DPUs. Unfortunately, despite the higher total number of DPUs, the caching still failed with out-of-memory errors. The core issue was that I had initially focused on the total DPU count rather than considering the instance type (G.8X vs. G.4X) and its impact on memory allocation per executor.

Why Caching Failed:

In the G.4X configuration:

• Reduced Memory per Executor: Each executor now had only 64 GB of memory, half of what was available with G.8X DPUs. This reduction meant that the same DataFrame, which could previously be cached in memory, now exceeded the available memory per executor.
• Increased Executor Count: Although the total DPU count was the same, having more executors (due to the higher number of DPUs) increased the overhead and complexity of managing data partitions. This also led to more frequent disk I/O as memory limits were breached.
• Memory Fragmentation: The smaller memory size per executor made it more challenging to handle large, contiguous blocks of data, leading to inefficiencies and eventual memory errors.

Even after increasing the DPUs to 120 and 160, the underlying issue of limited memory per executor remained, and the problem persisted.

When Lower Instance Sizes Are Beneficial:

• High Parallelism with Simple Tasks: Lower instance sizes, like G.2X,G.4X can be beneficial when you need high parallelism for tasks that are not memory-intensive. This allows you to distribute the workload across more executors, which can lead to faster processing times for operations that require less memory per executor.
• Cost Efficiency: If your workload does not require large memory per executor, using smaller instance sizes can be more cost-effective, allowing you to maximize resource usage without paying for more memory than you need.

When Higher Instance Sizes Are Beneficial:

• Memory-Intensive Operations: For tasks like caching large DataFrames or performing complex transformations that require significant memory, higher instance sizes, like G.8X are more suitable. They provide more memory per executor, reducing the risk of out-of-memory errors and allowing you to handle larger datasets in memory.
• Reduced I/O Overhead: With more memory per executor, you can reduce the need for disk I/O, which can significantly improve performance, especially for large-scale data processing tasks.

Lessons Learned:

1. Instance Type Matters: Simply increasing the number of DPUs is not always the solution. The type of DPU (G.8X vs. G.4X) significantly impacts how resources are allocated and how efficiently Spark can perform memory-intensive operations like caching.
2. Memory per Executor is Crucial: For tasks that involve large DataFrames or require extensive in-memory operations, having more memory per executor (as with G.8X) is often more beneficial than increasing the number of executors.
3. Repartitioning and Persistence: When dealing with large datasets that exceed memory limits, consider using persist(StorageLevel.MEMORY_AND_DISK) instead of cache(). Additionally, repartitioning data can help distribute the workload more evenly, though it might not always be enough if memory is the primary constraint.
4. Monitor and Adjust: Regularly monitor your Spark jobs via the Spark UI to understand where memory is being consumed. Adjust configurations like spark.executor.memory and spark.sql.shuffle.partitions based on the specific needs of your workload.
5. Consider Writing Intermediate Data: When caching or persisting fails due to memory limitations, an alternative is to write intermediate data to disk (e.g., S3) and read it back. While this increases I/O overhead, it can prevent memory-related failures.

Conclusion:

Data engineering in distributed environments is as much about understanding the tools and configurations as it is about the data itself. My experience highlighted how crucial it is to align your Spark and DPU configurations with the specific requirements of your job. Simply increasing the number of DPUs without considering the instance type and memory allocation per executor can lead to inefficiencies and even failures. By carefully considering memory allocation and resource management, you can optimize performance and avoid common pitfalls like out-of-memory errors. I hope these learnings help others navigate similar challenges in their data engineering journeys.

This experience serves as a reminder that the underlying architecture of your compute resources can have a profound impact on the success of your data processing tasks. As always, continuous learning and adaptation are key in the ever-evolving field of data engineering.