Flink memory usage

[AndreaKinn]

Hi,
I would like to share some considerations about Flink memory consumption.
I have a cluster composed of three nodes: 1 used both as JM and TM and other
2 TM.

I ran two identical applications (in different moments) on it. The only
difference is that on the second one I doubled every operators, essentially
to check what changes in resource's usage.

Analysing the outcomes on cpu side effectively the efforts are doubled.
Doing the same with memory I had these results:

<http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/file/t985/Screen_Shot_2017-11-03_at_21.png>

which to me seems completely counterintuitive since the results are
essentially equal.
I can imagine in the second case the memory was effectively almost full but
why Flink gets such a lot of memory even in the first case?
How it is explained this behaviour?



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

[Kien Truong]

Hi,

How did you measure the memory usage ?

JVM processes tend to occupy the maximum memory allocated to them, regardless of whether those memory are actively in used or not. To correctly measure the memory usage, you should use Flink's metric system[1]

Regards,

Kien

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.3/monitoring/metrics.html#system-metrics

On 11/4/2017 3:15 AM, AndreaKinn wrote:

Hi,
I would like to share some considerations about Flink memory consumption.
I have a cluster composed of three nodes: 1 used both as JM and TM and other
2 TM.

I ran two identical applications (in different moments) on it. The only
difference is that on the second one I doubled every operators, essentially
to check what changes in resource's usage.

Analysing the outcomes on cpu side effectively the efforts are doubled.
Doing the same with memory I had these results:

<http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/file/t985/Screen_Shot_2017-11-03_at_21.png> 

which to me seems completely counterintuitive since the results are
essentially equal.
I can imagine in the second case the memory was effectively almost full but
why Flink gets such a lot of memory even in the first case?
How it is explained this behaviour?



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

[AndreaKinn]

I have used sysstat linux tool.

On the node the only one application running is Flink. The outcomes measured
with metric system could be different?




--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

[AndreaKinn]

Anyway, If I understood how system metrics works (the results seems to be
showed in browser) I can't use it because my cluster is accessible only with
terminal via ssh



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

[Jürgen Thomann]

Can you use wget (curl will work as well)? You can find the taskmanagers
with wget -O - http://localhost:8081/taskmanagers
and wget -O - http://localhost:8081/taskmanagers/<id from previous
request> to see detailed jvm
memory stats. localhost:8081 is in my example the jobmanager.


On 04.11.2017 16:19, AndreaKinn wrote:
> Anyway, If I understood how system metrics works (the results seems to be
> showed in browser) I can't use it because my cluster is accessible only with
> terminal via ssh
>
>
>
> --
> Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

[Greg Hogan]

I’ve used the following simple script to capture Flink metrics by running:
        python -u ./statsd_server.py 9020 > statsd_server.log


>>> flink-conf.yaml
metrics.reporters: statsd_reporter
metrics.reporter.statsd_reporter.class: org.apache.flink.metrics.statsd.StatsDReporter
metrics.reporter.statsd_reporter.host: <JobManager IP>
metrics.reporter.statsd_reporter.port: 9020


>>> statsd_server.py
#!/usr/bin/env python

import socket
import sys
import time

if len(sys.argv) < 2:
  print('Usage {} <port>'.format(sys.argv[0]))
  sys.exit(-1)

UDP_IP = ''
UDP_PORT = int(sys.argv[1])

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind((UDP_IP, UDP_PORT))

while True:
    data, addr = sock.recvfrom(4096)
    print('{:.6f} {}'.format(time.time(), data))

> On Nov 5, 2017, at 4:40 AM, Jürgen Thomann <[hidden email]> wrote:
>
> Can you use wget (curl will work as well)? You can find the taskmanagers with wget -O - http://localhost:8081/taskmanagers
> and wget -O - http://localhost:8081/taskmanagers/<id from previous request> to see detailed jvm
> memory stats. localhost:8081 is in my example the jobmanager.
>
>
> On 04.11.2017 16:19, AndreaKinn wrote:
>> Anyway, If I understood how system metrics works (the results seems to be
>> showed in browser) I can't use it because my cluster is accessible only with
>> terminal via ssh
>>
>>
>>
>> --
>> Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/