OOM Issue in Streaming with foreachBatch()

dzsuzs
New Contributor II

‎06-03-2024 08:56 AM

I have a stateless streaming application that uses foreachBatch. This function executes between 10-400 times each hour based on custom logic. 
 
The logic within foreachBatch includes: 
  1. collect() on very small DataFrames (a few megabytes) --> driver memory is more than 20GB so it shouldn't be an issue
  2. Caching DataFrames and then unpersisting them 
  3. Converting a single row to a DF
  4. Performing a cross join on a very small DataFrame 
  5. Various filtering operations 
  6. Writing the DataFrame to the target_table in append mode.
The driver memory usage gradually increases over a few days until it eventually hits a Driver Out of Memory (OOM) error.
 
  • When does Spark remove state from the driver metadata in a streaming application? Are there configurations to force more aggressive cleanup?
  • Could frequent calls to collect() on small DataFrames still cause driver OOM issues? What alternatives can I use?
  • Should I avoid caching DataFrames even if they are used multiple times within a microbatch? How can I optimize the caching strategy?
  • Are there specific configurations or practices to better manage driver metadata and prevent memory bloat?
The goal is to find a solution to manage and optimize driver memory usage effectively.
 
I look forward to your suggestions and insights on resolving this issue.