You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@storm.apache.org by Karthikeyan Muthukumarasamy <mk...@gmail.com> on 2014/04/23 17:40:45 UTC
Reducing network traffic for multiple groupby
Hi,
I use Trident for real-time aggregation for multiple time-buckets, like 5
min, 15 mins, 1 hour etc, write partial aggregates as batches into
corresponding mart tables in database.
In the EventTimeBucketer function, I just floor the event_time to the
corresponding 5 min (or 15 min or 1 hour) starting epoch.
Something like this: long bucket_fivemin = event_time - (event_time %
(1000*60*5));
The topology looks like below:
Stream mainStream = topology.newStream("
PartitionedTransactionalSpout", spout)
.each(new Fields("event_time"), new EventTimeBucketer(),
new Fields("bucket_fivemin", "bucket_fifteenmin", "bucket_onehour"))
.parallelismHint(NUM_OF_PARTITIONS);
Stream fiveMinAggr = mainStream.
.groupBy(new Fields("bucket_fivemin",
"dimension1","dimension2", "dimension3, "dimension4"))
.chainedAgg()
.aggregate(new Fields("measure1"), new Sum(), new
Fields("measure1_sum"))
.aggregate(new Fields("measure2"), new Sum(), new
Fields("measure2_sum"))
.chainEnd()
.partitionAggregate(new Fields("bucket_fivemin", "dimension1",
"dimension2", "dimension3", "dimension4", "measure1_sum", "measure2_sum"),
new AggrPersistor(FIVE_MIN_AGGR), new Fields("done"))
.parallelismHint(10);
Stream fifteenMinAggr = mainStream.
.groupBy(new Fields("bucket_fifteenmin",
"dimension1","dimension2", "dimension3, "dimension4"))
.chainedAgg()
.aggregate(new Fields("measure1"), new Sum(), new
Fields("measure1_sum"))
.aggregate(new Fields("measure2"), new Sum(), new
Fields("measure2_sum"))
.chainEnd()
.partitionAggregate(new Fields("bucket_fifteenmin",
"dimension1", "dimension2", "dimension3", "dimension4", "measure1_sum",
"measure2_sum"),
new AggrPersistor(FIFTEEN_MIN_AGGR), new Fields("done"))
.parallelismHint(10);
and so on for one hour, one day etc.
Left as it is, a single event record, will now flow N times across the
network (once for each groupby combination), which would result in undue
network traffic.
How can I optimize network traffic for this usecase? Is it possible to use
Storm's pluggable scheduler feature to co-locate all bolts handling a
single combination of dimensions in a single machine to reduce network.
Any suggestions or code snippets will be very useful.
Thanks
MK
Re: Monitoring a Storm Cluster
Posted by Danijel Schiavuzzi <da...@schiavuzzi.com>.
You could try Monit, it's a process supervisor with additional system
monitoring and alerting features. It can alert you by e-mail on Supervisor
and Nimbus crashes and any other metrics you setup it to monitor. I believe
you could configure it to monitor Storm's Worker processes too.
On Fri, Apr 25, 2014 at 11:07 AM, Dimitris Amaxilatis <
d.amaxilatis@gmail.com> wrote:
> Hello everyone,
>
> I wanted to ask if there any configuration for a Strom cluster in order to
> get notifications (via mail ?) when a worker node goes down (or the
> connection the zookeeper ? ).
>
> I have been using storm for quite a while and managed to make my
> production topology quite stable but still have some minor craches on the
> worker nodes at some times.
>
> Checking the Storm UI is a solution but does not offer immediate
> notifications by itself.
>
> Thanks!
> Dimitris
>
--
Danijel Schiavuzzi
E: danijel@schiavuzzi.com
W: www.schiavuzzi.com
T: +385989035562
Skype: danijels7
Re: Monitoring a Storm Cluster
Posted by Otis Gospodnetic <ot...@gmail.com>.
Hi,
Not an open-source solution, but you can get "heartbeat alerts" when your
worker node goes down if you use SPM <http://sematext.com/spm/>. If that's
the main thing you are after, pick a free plan and you're good to go.
Otis
--
Performance Monitoring * Log Analytics * Search Analytics
Solr & Elasticsearch Support * http://sematext.com/
On Fri, Apr 25, 2014 at 5:07 AM, Dimitris Amaxilatis <d.amaxilatis@gmail.com
> wrote:
> Hello everyone,
>
> I wanted to ask if there any configuration for a Strom cluster in order to
> get notifications (via mail ?) when a worker node goes down (or the
> connection the zookeeper ? ).
>
> I have been using storm for quite a while and managed to make my
> production topology quite stable but still have some minor craches on the
> worker nodes at some times.
>
> Checking the Storm UI is a solution but does not offer immediate
> notifications by itself.
>
> Thanks!
> Dimitris
>
Monitoring a Storm Cluster
Posted by Dimitris Amaxilatis <d....@gmail.com>.
Hello everyone,
I wanted to ask if there any configuration for a Strom cluster in order
to get notifications (via mail ?) when a worker node goes down (or the
connection the zookeeper ? ).
I have been using storm for quite a while and managed to make my
production topology quite stable but still have some minor craches on
the worker nodes at some times.
Checking the Storm UI is a solution but does not offer immediate
notifications by itself.
Thanks!
Dimitris
Re: Reducing network traffic for multiple groupby
Posted by Lorenz Fischer <lo...@gmail.com>.
Hi MK
You may want to have a look at this [1] post or the code for this
paper [2] which you can find here [3]. I'm currently working on an
implementation of an idea we presented in this paper [4] (a short
video of the idea can be found here [5]).
Cheers,
Lorenz
[1] http://xumingming.sinaapp.com/885/twitter-storm-how-to-develop-a-pluggable-scheduler/
[2] http://dl.acm.org/citation.cfm?id=2488267
[3] http://www.dis.uniroma1.it/~midlab/software/storm-adaptive-schedulers.zip
[4] http://www.merlin.uzh.ch/publication/show/8337
[5] https://www.youtube.com/watch?v=dLLOKo9VRag
On Wed, Apr 23, 2014 at 5:40 PM, Karthikeyan Muthukumarasamy
<mk...@gmail.com> wrote:
> Hi,
> I use Trident for real-time aggregation for multiple time-buckets, like 5
> min, 15 mins, 1 hour etc, write partial aggregates as batches into
> corresponding mart tables in database.
> In the EventTimeBucketer function, I just floor the event_time to the
> corresponding 5 min (or 15 min or 1 hour) starting epoch.
> Something like this: long bucket_fivemin = event_time - (event_time %
> (1000*60*5));
>
> The topology looks like below:
>
> Stream mainStream = topology.newStream("
> PartitionedTransactionalSpout", spout)
> .each(new Fields("event_time"), new EventTimeBucketer(), new
> Fields("bucket_fivemin", "bucket_fifteenmin", "bucket_onehour"))
> .parallelismHint(NUM_OF_PARTITIONS);
>
> Stream fiveMinAggr = mainStream.
> .groupBy(new Fields("bucket_fivemin", "dimension1","dimension2",
> "dimension3, "dimension4"))
> .chainedAgg()
> .aggregate(new Fields("measure1"), new Sum(), new
> Fields("measure1_sum"))
> .aggregate(new Fields("measure2"), new Sum(), new
> Fields("measure2_sum"))
> .chainEnd()
> .partitionAggregate(new Fields("bucket_fivemin", "dimension1",
> "dimension2", "dimension3", "dimension4", "measure1_sum", "measure2_sum"),
> new AggrPersistor(FIVE_MIN_AGGR), new Fields("done"))
> .parallelismHint(10);
>
> Stream fifteenMinAggr = mainStream.
> .groupBy(new Fields("bucket_fifteenmin",
> "dimension1","dimension2", "dimension3, "dimension4"))
> .chainedAgg()
> .aggregate(new Fields("measure1"), new Sum(), new
> Fields("measure1_sum"))
> .aggregate(new Fields("measure2"), new Sum(), new
> Fields("measure2_sum"))
> .chainEnd()
> .partitionAggregate(new Fields("bucket_fifteenmin",
> "dimension1", "dimension2", "dimension3", "dimension4", "measure1_sum",
> "measure2_sum"),
> new AggrPersistor(FIFTEEN_MIN_AGGR), new Fields("done"))
> .parallelismHint(10);
>
> and so on for one hour, one day etc.
>
> Left as it is, a single event record, will now flow N times across the
> network (once for each groupby combination), which would result in undue
> network traffic.
> How can I optimize network traffic for this usecase? Is it possible to use
> Storm's pluggable scheduler feature to co-locate all bolts handling a single
> combination of dimensions in a single machine to reduce network.
> Any suggestions or code snippets will be very useful.
>
> Thanks
> MK