You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Zsuzsanna D <dz...@gmail.com> on 2024/06/03 14:50:04 UTC
OOM Issue in Streaming with foreachBatch()
Hi everyone,
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.
Best,
Zs
Re: OOM Issue in Streaming with foreachBatch()
Posted by Khalid Mammadov <kh...@gmail.com>.
If your memory increases gradually to cause OOM I would suggest looking at
max memory settings of driver as Java eagerly allocates memory if it's
allowed and does not release unless max is set correctly to cause GC and
keep it under available memory of the VM it runs on.
On Mon, 3 Jun 2024, 15:50 Zsuzsanna D, <dz...@gmail.com> wrote:
> Hi everyone,
>
> 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.
>
> Best,
> Zs
>
Re: OOM Issue in Streaming with foreachBatch()
Posted by Zsuzsanna D <dz...@gmail.com>.
I do have method calls from the class but they are not listeners, and I
have no dependency or data exchange between mico-batches.
Mich Talebzadeh <mi...@gmail.com> ezt írta (időpont: 2024. jún.
4., K, 15:20):
> The core concept of a stateless application in Spark Structured Streaming
> is that each microbatch is processed independently, without relying on
> information from previous batches. Do you have the code similar to below
>
> def sendToSink(df, batchId):
> if(len(df.take(1))) > 0:
> print(f"""From sendToSink, md, batchId is {batchId}, at
> {datetime.now()} """)
> df.persist()
> # Call the method from the class
> PrintClass().printMethod(df)
> #print(MyListener().onQueryStarted(StreamingQueryListener))
> #print(MyListener().onQueryProgress(StreamingQueryListener))
> # write to BigQuery batch table
> #s.writeTableToBQ(df, "append",
> config['MDVariables']['targetDataset'],config['MDVariables']['targetTable'])
> df.unpersist()
> #print(f"""wrote to DB""")
> else:
> print("DataFrame md is empty")
>
> HTH
>
> Mich Talebzadeh,
> Technologist | Architect | Data Engineer | Generative AI | FinCrime
> PhD <https://en.wikipedia.org/wiki/Doctor_of_Philosophy> Imperial College
> London <https://en.wikipedia.org/wiki/Imperial_College_London>
> London, United Kingdom
>
>
> view my Linkedin profile
> <https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
>
>
> https://en.everybodywiki.com/Mich_Talebzadeh
>
>
>
> *Disclaimer:* The information provided is correct to the best of my
> knowledge but of course cannot be guaranteed . It is essential to note
> that, as with any advice, quote "one test result is worth one-thousand
> expert opinions (Werner <https://en.wikipedia.org/wiki/Wernher_von_Braun>Von
> Braun <https://en.wikipedia.org/wiki/Wernher_von_Braun>)".
>
>
> On Tue, 4 Jun 2024 at 00:46, Zsuzsanna D <dz...@gmail.com> wrote:
>
>> Hi everyone,
>>
>> 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.
>>
>> Best,
>> Zs
>>
>
Re: OOM Issue in Streaming with foreachBatch()
Posted by Mich Talebzadeh <mi...@gmail.com>.
The core concept of a stateless application in Spark Structured Streaming
is that each microbatch is processed independently, without relying on
information from previous batches. Do you have the code similar to below
def sendToSink(df, batchId):
if(len(df.take(1))) > 0:
print(f"""From sendToSink, md, batchId is {batchId}, at
{datetime.now()} """)
df.persist()
# Call the method from the class
PrintClass().printMethod(df)
#print(MyListener().onQueryStarted(StreamingQueryListener))
#print(MyListener().onQueryProgress(StreamingQueryListener))
# write to BigQuery batch table
#s.writeTableToBQ(df, "append",
config['MDVariables']['targetDataset'],config['MDVariables']['targetTable'])
df.unpersist()
#print(f"""wrote to DB""")
else:
print("DataFrame md is empty")
HTH
Mich Talebzadeh,
Technologist | Architect | Data Engineer | Generative AI | FinCrime
PhD <https://en.wikipedia.org/wiki/Doctor_of_Philosophy> Imperial College
London <https://en.wikipedia.org/wiki/Imperial_College_London>
London, United Kingdom
view my Linkedin profile
<https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
https://en.everybodywiki.com/Mich_Talebzadeh
*Disclaimer:* The information provided is correct to the best of my
knowledge but of course cannot be guaranteed . It is essential to note
that, as with any advice, quote "one test result is worth one-thousand
expert opinions (Werner <https://en.wikipedia.org/wiki/Wernher_von_Braun>Von
Braun <https://en.wikipedia.org/wiki/Wernher_von_Braun>)".
On Tue, 4 Jun 2024 at 00:46, Zsuzsanna D <dz...@gmail.com> wrote:
> Hi everyone,
>
> 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.
>
> Best,
> Zs
>