Hi Kishore, This is an excellent response--thanks for taking time to write back! This is great analysis of the problem space. I/O overhead--disk plus network--is a big problem for us.
The suggestion to use Local grouping is an excellent one, but I need to use the Fields grouping at some point--either from the KafkaSpout to Bolt A, or Bolt A to Bolt B. The reason I need to do this is we're caching Key/Value pairs and then persisting each collection to a SequenceFile once each cache collection reaches a certain size. I currently have fields grouping for Bolt A to Bolt B, and would like to maintain this approach as the KafkaSpout has 10 partitions as input, any one of which can develop hotspots. Consequently, I am using the Shuffle grouping for the KafkaSpout to Bolt A step to evenly distribute the tuples coming from Kafka to Bolt A. I then use Fields grouping to ensure the same Key/Value pairs go tot the same Bolt B instance. However, as you point out, the network overhead is really throttling throughput, so it's worth a shot to do the fields grouping at the KafkaSpout-Bolt A step and then do a Local grouping at the Bolt A-Bolt B step. Since I have one worker per node as I mentioned in a response to Javier's earlier post, I will need to up the number of Bolt B instances to match Bolt A to ensure Local shuffling for all tuples. Not totally sure how much adding 300 additional executors for 120 workers will negatively impact performance, but I am guessing that it would be more than offset by the greatly decreased network and disk I/O, since Bolt A-Bolt B will be all intra-Java process messaging via IMAX. Thanks again for your excellent thoughts! --John On Fri, Aug 14, 2015 at 9:48 PM, Kishore Senji <[email protected]> wrote: > 7 million/minute with 200 instances, implies a latency > of 200*1000/(7*1000^2/60) ~ 1.71 ms for bolt A. For throughput > calculations, I would normalize it and visualize a single instance of the > bolt running. So for 1 both instance, the latency would be 1.71/200. > > By adding bolt B with 50 instances, the throughput came to 900000/min, > which means > > 900000= 1000*60 / ((1.71/200)+(B/50)), solving this B would be around 2.9 > ms. Even though you mentioned there is nothing in the Bolt B impl, the > overall latency is 2.9ms, this means there is a lot of overhead in the > network. Are you using fields grouping or because of the number of > instances the tuples are sent over the network. Try using local grouping. > > Bumping bolt B to 100 instances, you got 1.1 million/minute. But this > equation should give us: 1000*60 / ((1.71/200)+(2.9/100)) around 1.6 > million/min. This means adding more instances added more overhead, could be > that more number of tuples are going over the network (because they went in > to different worker processes). Calculating the bolt b latency for 1.1 > million gives = 4.6 ms of overhead for Bolt B. > > Adding more instances can be costly because of the IPC. The key would be > to get more local shuffling. Check if you have any failures in the overall > system and also check the Storm UI which will show the bottleneck in your > topology (the capacity of the bolts and the latencies etc) > > > On Fri, Aug 14, 2015 at 2:43 PM Javier Gonzalez <[email protected]> > wrote: > >> Do you actually have 170 machines? Try sticking to one worker per >> machine (tweak memory parameters in storm.yaml), makes inter bolt traffic >> much faster. >> On Aug 14, 2015 5:28 PM, "John Yost" <[email protected]> wrote: >> >>> Hey Javier, >>> >>> Cool, thanks for your response! I have 50 workers for 200 Bolt A/5 Bolt >>> B and 120 workers for 400 Bolt A/100 Bolt B (this latter config is optimal, >>> but cluster resources make it tricky to actually launch this). >>> >>> I will up the number of Ackers and see if that helps. If not, then I >>> will try to vary the number of B bolts beyond 100. >>> >>> Thanks Again! >>> >>> --John >>> >>> On Fri, Aug 14, 2015 at 2:59 PM, Javier Gonzalez <[email protected]> >>> wrote: >>> >>>> You will have a detrimental effect to wiring in boltB, even if it does >>>> nothing but ack. Every tuple you have processed from A has to travel to a B >>>> bolt, and the ack has to travel back. >>>> >>>> You could try modifying the number of ackers, and playing with the >>>> number of A and B bolts. How many workers do you have for the topology? >>>> >>>> Regards, >>>> JG >>>> On Aug 14, 2015 12:31 PM, "John Yost" <[email protected]> >>>> wrote: >>>> >>>>> Hi Everyone, >>>>> >>>>> I have a topology where a highly CPU-intensive bolt (Bolt A) requires >>>>> a much higher degree of parallelism than the bolt it emits tuples to (Bolt >>>>> B) (200 Bolt A executors vs <= 100 Bolt B executors). >>>>> >>>>> I find that the throughput, as measured in number of tuples acked, >>>>> goes from 7 million/minute to ~ 1 million/minute when I wire in Bolt >>>>> B--even if all of the logic within the Bolt B execute method is disabled >>>>> and the Bolt B is therefore simply acking the input tuples from Bolt A. In >>>>> addition, I find that, going from 50 to 100 Bolt B executors causes the >>>>> throughput to go from 900K/minute to ~ 1.1 million/minute. >>>>> >>>>> Is the fact that I am going from 200 bolt instances to 100 or less the >>>>> problem? I've already experimented with executor.send.buffer.size and >>>>> executor.receive.buffer.size, which helped drive throughput from 800K to >>>>> 900K. I will try topology.transfer.buffer.size, perhaps set that higher to >>>>> 2048. Any other ideas? >>>>> >>>>> Thanks >>>>> >>>>> --John >>>>> >>>>> >>>
