Hi Stefan, The throughput was the same in all of the executions. This was well validated in each test, as this is what I also suspected that can be effected. The throughput was ~110,000 messages per second. Regarding the code example, this is a bit confidential, let me think what I can do and get back to you. Am I the first one who encountered such an issue?
Thanks, Liron From: Stefan Richter [mailto:s.rich...@data-artisans.com] Sent: Thursday, January 04, 2018 11:15 AM To: Netzer, Liron [ICG-IT] Cc: user@flink.apache.org Subject: Re: Lower Parallelism derives better latency Hi, ok that would have been good to know, so forget about my explanation attempt :-). This makes it interesting, and at the same time I cannot come up with an „easy“ explanation. It is not even clear if the reason for this is a general problem in Flink, your setup, or caused by something that your job is doing. Two more questions: What happens to the throughput in that experiment? Does it also decrease or increase? I just want to rule out that some general overhead is introduced. Second, do you have or could you create some (minimal) code example to reproduce the problem that you could share with us (of course you can also share this in privat)? This would be very helpful! Best, Stefan Am 04.01.2018 um 08:45 schrieb Netzer, Liron <liron.net...@citi.com<mailto:liron.net...@citi.com>>: Hi Stefan, Thanks for replying. All of the tests below were executed with a buffer timeout of zero: env.setBufferTimeout(0); so this means that the buffers were flushed after each record. Any other explanation? ☺ Thanks, Liron From: Stefan Richter [mailto:s.rich...@data-artisans.com] Sent: Wednesday, January 03, 2018 3:20 PM To: Netzer, Liron [ICG-IT] Cc: user@flink.apache.org<mailto:user@flink.apache.org> Subject: Re: Lower Parallelism derives better latency Hi, one possible explanation that I see is the following: in a shuffle, each there are input and output buffers for each parallel subtask to which data could be shuffled. Those buffers are flushed either when full or after a timeout interval. If you increase the parallelism, there are more buffers and each buffer gets a smaller fraction of the data. This, in turn, means that it takes longer until an individual buffer is full and data is emitted. The timeout interval enforces an upper bound. Your experiments works on a very small scale, and I would not assume that this would increase latency without bounds - at least once you hit the buffer timeout interval the latency should no longer increase. You could validate this by configuring smaller buffer sizes and test how this impacts the experiment. Best, Stefan Am 03.01.2018 um 08:13 schrieb Netzer, Liron <liron.net...@citi.com<mailto:liron.net...@citi.com>>: Hi group, We have a standalone Flink cluster that is running on a UNIX host with 40 CPUs and 256GB RAM. There is one task manager and 24 slots were defined. When we decrease the parallelism of the Stream graph operators(each operator has the same parallelism), we see a consistent change in the latency, it gets better: Test run Parallelism 99 percentile 95 percentile 75 percentile Mean #1 8 4.15 ms 2.02 ms 0.22 ms 0.42 ms #2 7 3.6 ms 1.68 ms 0.14 ms 0.34 ms #3 6 3 ms 1.4 ms 0.13 ms 0.3 ms #4 5 2.1 ms 0.95 ms 0.1 ms 0.22 ms #5 4 1.5 ms 0.64 ms 0.09 ms 0.16 ms This was a surprise for us, as we expected that higher parallelism will derive better latency. Could you try to assist us to understand this behavior? I know that when there are more threads that are involved, there is probably more serialization/deserialization, but this can't be the only reason for this behavior. We have two Kafka sources, and the rest of the operators are fixed windows, flatmaps, coMappers and several KeyBys. Except for the Kafka sources and some internal logging, there is no other I/O (i.e. we do not connect to any external DB/queue) We use Flink 1.3. Thanks, Liron
