Flink application does not scale as expected, please help!
Locked
Flink application does not scale as expected, please help!
 
|
Hi,
We found that our Flink application with simple logic, which using process function is not scale-able when scale from 8 parallelism onward even though with sufficient resources. Below it the result which is capped at ~250k TPS. No matter how we tune the parallelism of the operators it just not scale, same to increase source parallelism.
Please refer to "scaleNotWork.png",
Sample source code attached(flink_app_parser_git.zip). It is a simple program, parsing json record into object, and pass it to a empty logic Flink's process function. Rocksdb is in used, and the source is generated by the program itself. This could be reproduce
easily.
We choose Flink because of it scalability, but this is not the case now, appreciated if anyone could help as this is impacting our projects! thank you.
To run the program, sample parameters, "aggrinterval=6000000 loop=7500000 statsd=1 psrc=4 pJ2R=32 pAggr=72 URL=do36.mycompany.com:8127"
We are running in VMWare, 5 Task Managers and each has 32 slots.
total used free shared buff/cache available
Please refer TM.png and JM.png for further details.
Thanks.
Regards, Yow
 flink_app_parser_git.zip (108K) Download Attachment JM.png (143K) Download Attachment sample.png (72K) Download Attachment scaleNotWork.png (109K) Download Attachment TM.png (234K) Download Attachment |
Re: Flink application does not scale as expected, please help!
|
|
Re: Flink application does not scale as expected, please help!
|
|
Re: Flink application does not scale as expected, please help!
|
|
Re: Flink application does not scale as expected, please help!
|
|
Re: Flink application does not scale as expected, please help!
|
|
Re: Flink application does not scale as expected, please help!
|
|
Re: Flink application does not scale as expected, please help!
|
|
Re: Flink application does not scale as expected, please help!
|
|
Re: Flink application does not scale as expected, please help!
|
|
Re: Flink application does not scale as expected, please help!
|
|
Re: Flink application does not scale as expected, please help!
|
|
Thanks @Fabian for your confirmation about the better performance when scaling happened at same TM machine. But it is so funny that it give impression "the more I scale the less I get" when the performance drop with more
TM in play. @Ovidiu question is interesting to know too. @Till do you mind to share your thoughts?
Thank you guys!
From: Ovidiu-Cristian MARCU <[hidden email]>
Sent: Monday, June 18, 2018 6:28 PM To: Fabian Hueske Cc: Siew Wai Yow; Jörn Franke; [hidden email] Subject: Re: Flink application does not scale as expected, please help! Hi all,
Allow me to add some comments/questions on this issue that is very interesting.
According to documentation [1] the pipeline example assumes the source is running with the same parallelism as successive map operator and the workflow optimizes to collocate source and map tasks if possible.
For an application configuring the source with different parallelism, assuming N task managers each with m slots, if I configure
the source operator with parallelism m, then all of the source's tasks could be scheduled on the first task manager?
I think the same story holds for sinks tasks.
So, in general is there any control over scheduling of source and sink tasks?
Would it be possible to enforce scheduling of source tasks to be balanced across task managers? Not sure if this is the default.
If the source creates a non-keyed stream, can we enforce the source to push records to local map tasks?
For Siew’s example, after source#map a keyBy complicates further things since each key can be possibly processed on another task manager.
At least the keyBy operator should run with the same parallelism as source and map and be pipelined on same slot (maybe shared slot configuration could enforce that).
DataStream<Record> AggregatedRecordWithAuditStream = sourceStringStream
.map(new JsonToRecordTranslator(markerFactory.getMarker(), inputlink)).name("JsonRecTranslator").setParallelism(pJ2R)
.keyBy(new KeySelector<Record, String>() {
private static final long serialVersionUID = 1L;
@Override
public String getKey(Record r) throws Exception {
return r.getUNIQUE_KEY();
}
})
.process(new ProcessAggregation(aggrDuration, markerFactory.getMarker(), markerFactory.getMarker())).setParallelism(pAggr)
.name("AggregationDuration: " + aggrDuration +"ms");
Thanks,
Ovidiu
|
|||
Re: Flink application does not scale as expected, please help!
|
|
Hi, as Fabian explained, if you exceed the number of slots on a single TM, then Flink needs to deploy tasks on other TMs which causes a network transfer between the sources and the mappers. This will have a negative impact if you compare it to a setup where everything runs locally. The scheduling in Flink works as follows: Flink will try to co-locate consumer tasks with their producers in order to avoid costly network communication. If there are more than 8 producer tasks, then it is assumed that the consumer can be scheduled everywhere because you might only achieve to read 1/8 of your input locally. The question now is how the sources are scheduled, because they determine how the mappers are scheduled. For tasks with no inputs, Flink will pick a random slot without preferences for spreading them across as many TMs as possible or packing them as closely together as possible. The reason is that both strategies might perform better depending on the actual use case. Moreover, the way the scheduling works right now, the scheduler does not know how many sources will be scheduled. This means that you don't necessarily know how many TMs you have to start in order to execute all source tasks and, thus, you don't know across how many TMs you have to spread your sources. Due to the implementation of the SlotManager, which is responsible for the slot management on the RM, we will first allocate all slots from a single TM before allocating slots from another TM. Due to the setup of your test (different parallelism of sources and mappers), you introduce a shuffle between the sources and the Json mappers. This is not optimal and I would suggest to instantiate as many sources as you have mappers. That way these two operators can be chained and run in the same thread. This would also spread the load of generating events across the whole cluster instead of letting a single machine be responsible for producing all events. Moreover, in order to get rid of the measurement artifacts introduced by switching from local to remote communication you could start your TMs with a single slot as Fabian suggested. Otherwise you might see a slight drop whenever you exceed the multiple of 16 by one. As a side node, if you don't need a shuffle when changing the parallelism between two operators, then you could also use the rescale command which tries to minimize the communication between two operators. For example, if your source has a parallelism of 2 and your mapper of 4, then rescale would cause the first source to talk to mapper 1 and 2 and the second source to only talk to mapper 3 and 4. Cheers, Till On Mon, Jun 18, 2018 at 5:30 PM Siew Wai Yow <[hidden email]> wrote:
|
|||
Re: Flink application does not scale as expected, please help!
|
|
In reply to this post by Siew Wai Yow
Hi Siew, The hint about the lower source parallelism compared to the operator parallelism might be the right one. Can you check if all source tasks are scheduled to the same machine? In that case your application might be bottlenecked by the out-going network connection of that single machine. Best, Fabian 2018-06-18 17:29 GMT+02:00 Siew Wai Yow <[hidden email]>:
|
|||
Re: Flink application does not scale as expected, please help!
|
|
In reply to this post by Till Rohrmann
Thanks @Fabian and @Till for the good explanation. The picture is pretty clear right now. As for the single slot TM test, seems it trying to allocate the same machine's slot first as well. But with that the result is less spiky. i guess @Fabian is right that network is our bottleneck in case because we have only 1Gbps. Will fix it and test again.
Thanks guys, including @Sihua zhou who help to test the program in his lab :)
From: Till Rohrmann <[hidden email]>
Sent: Tuesday, June 19, 2018 3:48 PM To: [hidden email] Cc: [hidden email]; Fabian Hueske; Jörn Franke; user Subject: Re: Flink application does not scale as expected, please help! Hi,
as Fabian explained, if you exceed the number of slots on a single TM, then Flink needs to deploy tasks on other TMs which causes a network transfer between the sources and the mappers. This will have a negative impact if you compare it to a setup where
everything runs locally.
The scheduling in Flink works as follows: Flink will try to co-locate consumer tasks with their producers in order to avoid costly network communication. If there are more than 8 producer tasks, then it is assumed that the consumer can be scheduled everywhere
because you might only achieve to read 1/8 of your input locally.
The question now is how the sources are scheduled, because they determine how the mappers are scheduled. For tasks with no inputs, Flink will pick a random slot without preferences for spreading them across as many TMs as possible or packing them as closely
together as possible. The reason is that both strategies might perform better depending on the actual use case. Moreover, the way the scheduling works right now, the scheduler does not know how many sources will be scheduled. This means that you don't necessarily
know how many TMs you have to start in order to execute all source tasks and, thus, you don't know across how many TMs you have to spread your sources. Due to the implementation of the SlotManager, which is responsible for the slot management on the RM, we
will first allocate all slots from a single TM before allocating slots from another TM.
Due to the setup of your test (different parallelism of sources and mappers), you introduce a shuffle between the sources and the Json mappers. This is not optimal and I would suggest to instantiate as many sources as you have mappers. That way these two
operators can be chained and run in the same thread. This would also spread the load of generating events across the whole cluster instead of letting a single machine be responsible for producing all events. Moreover, in order to get rid of the measurement
artifacts introduced by switching from local to remote communication you could start your TMs with a single slot as Fabian suggested. Otherwise you might see a slight drop whenever you exceed the multiple of 16 by one.
As a side node, if you don't need a shuffle when changing the parallelism between two operators, then you could also use the rescale command which tries to minimize the communication between two operators. For example, if your source has a parallelism
of 2 and your mapper of 4, then rescale would cause the first source to talk to mapper 1 and 2 and the second source to only talk to mapper 3 and 4.
Cheers,
Till
From: Fabian Hueske <[hidden email]>
Sent: Tuesday, June 19, 2018 3:55 PM To: Siew Wai Yow Cc: Ovidiu-Cristian MARCU; Jörn Franke; [hidden email]; [hidden email] Subject: Re: Flink application does not scale as expected, please help! Hi Siew,
The hint about the lower source parallelism compared to the operator parallelism might be the right one.
Can you check if all source tasks are scheduled to the same machine?
In that case your application might be bottlenecked by the out-going network connection of that single machine.
Best, Fabian
On Mon, Jun 18, 2018 at 5:30 PM Siew Wai Yow <[hidden email]> wrote:
|
|||
| Free forum by Nabble | Edit this page |