HI David, I'm not familiar with Flink, but assuming there aren't any memory management issues in the runner or SDK, try reducing proceesing_time_duration (30 minutes currently) to 60 seconds and see how long it takes for memory usage to reach the limit. Could you also say how long it currently takes for memory to reach the limit?
On Thu, Aug 6, 2020 at 4:23 PM David Gogokhiya <david...@yelp.com> wrote: > Hi, > > We recently started using Apache Beam version 2.20.0 running on Flink > version 1.9 deployed on kubernetes to process unbounded streams of data. > However, we noticed that the memory consumed by stateful Beam is steadily > increasing over time with no drops no matter what the current bandwidth is. > We were wondering if this is expected and if not what would be the best way > to resolve it. > More Context > > We have the following pipeline that consumes messages from the unbounded > stream of data. Later we deduplicate the messages based on unique message > id using the deduplicate function > <https://beam.apache.org/releases/pydoc/2.22.0/_modules/apache_beam/transforms/deduplicate.html#DeduplicatePerKey>. > Since we are using Beam version 2.20.0, we copied the source code of the > deduplicate function > <https://beam.apache.org/releases/pydoc/2.22.0/_modules/apache_beam/transforms/deduplicate.html#DeduplicatePerKey> > from version 2.22.0. After that we unmap the tuple, retrieve the necessary > data from message payload and dump the corresponding data into the log. > > Pipeline: > > > Flink configuration: > > > As we mentioned before, we noticed that the memory usage of the jobmanager > and taskmanager pod are steadily increasing with no drops no matter what > the current bandwidth is. We tried allocating more memory but it seems like > no matter how much memory we allocate it eventually reaches its limit and > then it tries to restart itself. > > > Sincerely, David > >
smime.p7s
Description: S/MIME Cryptographic Signature
