Hi, does anyone have any tips on monitoring and troubleshooting PyFlink memory usage (in k8s)? We are running a simple datastream enrichment app. kafka input -> enrich via http api request -> kafka output. Over time our (single, for now) taskmanager is killed by k8s because it reaches its memory limit. All of the JVM metrics look normal. On the container, I can see the python process slowly using more and more RSS mem. We’re going to try to add instrumentation to get metrics about the python enrich ProcessFunctions, but we’re kind of at a loss for what to look into.

@Andrew Otto What are you doing in the Python ProcessFunctions? How long the job has run before getting killed?

This is the python

func

we are calling in the ProcessFunction. And this is the full ProcessFunction we apply the datastream to.

@Andrew Otto I’m a little surprised if the memory of Python process was keep growing as it seems that you just process the elements row by row and have not buffered elements. Could you try to adjust the following configurations to let the job running a little longer(e.g. one hour or so) to see if the memory will still grow after running one hour or so:

taskmanager.memory.managed.fraction: 0.4 (default)
taskmanager.memory.managed.consumer-weights: OPERATOR:70,STATE_BACKEND:70,PYTHON:30	(default)
jobmanager.memory.jvm-overhead.fraction: 0.1 (default)

okay thanks! Will give it a go. I’m working on getting some better memory and other metrics in now, and then we’ll try to do some more tuning. We are also going to try an async python http client, but i’m not really sure how that fits in with Flink, since PyFlink doesn’t itself support Async IO operators

We are also going to try an async python http client, but i’m not really sure how that fits in with Flink, since PyFlink doesn’t itself support Async IO operators

This is still not supported (We are planning to provide support for this in Flink 1.18). For now, if you could torrent losing some records during job failover, you could try to use async http client to retrieve record inside

FlatMapFunction

of

datastream.flat_map(FlatMapFunction)

, the

flat_map

of

FlatMapFunction

may look like the following: • perform async http client operations for the newly incoming record • check if the previous results are already available and return them if available The side effects of this approach: • During failover, in-complete records will be discarded

Hi 👋. I'm a colleague of @Andrew Otto working to troubleshoot this memory issue. @Dian Fu I experimented with the tunable you pointed us to, and was able to run the application for over 6 hours. Memory kept growing till container OOM. Some additional info that I was able to observe: 1. while memory consumption grows over time, the number of objects tracked by Python's GC seem to stabilize after a few minutes of runtime, and not grow linearly with memory. 2. How quickly memory grows seems to be proportional to

python.fn-execution.arrow.batch.size

@Gabriele Modena

while memory consumption grows over time, the number of objects tracked by Python’s GC seem to stabilize after a few minutes of runtime, and not grow linearly with memory.

Does the Python process memory continues grow in this case?

unfortunately it does. I had a run were it seemed to stabilize, but eventually kept growing till OOM a few hours later.

@Gabriele Modena I guess you could try the following: • Try to comment all of the code inside your

ProcessFunction

to check whether the memory leak comes from the PyFlink framework or your custom code • If it comes your custom code, try to comment out part of the logic to identify which part of code the memory leak comes from Besides, I guess you could also try tools like the following: https://web.archive.org/web/20160316152429/http://www.lshift.net/blog/2008/11/14/tracing-python-memory-leaks/

Try to comment all of the code inside your

ProcessFunction

to check whether the memory leak comes from the PyFlink framework or your custom code

That's the path I'm taking now, with some heuristics I got from cProfile. I'll report back if something interesting comes out.

@Dian Fu re async pyflink in Flink 1.18. Do you have a guess on when this might be available? 6 months? 2 years? 🙂

@Andrew Otto About 6 months. Flink 1.18 is planned to be feature freeze on end of June or Mid of July and released in about 1.5 ~ 2 month later.

Ok, thank you!