You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@storm.apache.org by Svend Vanderveken <sv...@gmail.com> on 2013/12/06 12:14:51 UTC
Re: Storm Scalability : Topology to process 5 million records
HI,
Side note for our non-indian friends: a "lac" is 100000 :D
Bishnu, it seems indeed your topology has read and processed 325000 storm
tuples, do you confirm the topology is using one storm tuple for each of
the 5 million records you mention, or are they regrouped somehow?
Also, note that Storm is usually designed to process "infinite" streams of
data, there's not really a concept of a start and an end of a stream.
Especially, the error-handling mechanism based on replay of tuples (see doc
on Opaque/transactional) imply that some tuples could be processed several
times, thus increasing the counts you see in the stats.
If you want to know when a certain number of different tuples have been
processed, the only way I can imagine is actually counting yourself the
number of distinct record ids you send to RabbitMQ (and make sure you do so
in a scalable manner).
Cheers
S
On Fri, Dec 6, 2013 at 11:56 AM, Bishnu Patro <Bi...@symantec.com>wrote:
> Hi,
>
>
>
> I have executed the topology with following configurations for 5 million
> records:
>
>
>
> *Topology configurations:*
>
>
>
> Number of spout: 1, Number of executor: 1
>
> Number of Bolt: 1, Number of executor: 20
>
> Number of workers: 4
>
> Number of acks: 2
>
> Number of spout pending: 20000
>
> Optimize flag: true
>
> Debug flag: false
>
>
>
> My topology is running on a single machine with
> zookeeper/nimbus/supervisor as different process.
>
> I am using Rabbit MQ for pushing messages which will be consumed by spout
> and output of bolt is pushed back to Rabbit MQ in a different queue.
>
>
>
> *Machine Configurations*
>
>
>
> RAM: 32 GB
>
> CPU: 8 core
>
> HDD: 500 GB
>
>
>
> After some time ( > 3hrs ) when I saw the topology summary (psb) in the
> storm UI, I don’t see any error, but number of emitted is showing only
> close to 6.5 lacs and number of acked is showing 3.25 lacs, none of
> messages has failed, although all 5 millions tuples are pushed to RabbitMQ.
>
>
>
> *Questions:*
>
>
>
> What is the reason for not showing 5 million messages and how can I
> conclude that all my messages are processed ?
>
> How to read the topology summary, If I want to find the time
> taken/throughput to execute 5 million tuples ?
>
>
>
> Please forgive for this long mail.
> Topology summary
>
> *Name*
>
> *Id*
>
> *Status*
>
> *Uptime*
>
> *Num workers*
>
> *Num executors*
>
> *Num tasks*
>
> Incidentstopology
>
> Incidentstopology-1-1386292945
>
> ACTIVE
>
> 3h 15m 44s
>
> 4
>
> 23
>
> 23
> Topology actionsTopology stats
>
> *Window*
>
> *Emitted*
>
> *Transferred*
>
> *Complete latency (ms)*
>
> *Acked*
>
> *Failed*
>
> 10m 0s<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945?window=600>
>
> 0
>
> 0
>
> 0.000
>
> 0
>
> 0
>
> 3h 0m 0s<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945?window=10800>
>
> 651880
>
> 326400
>
> 16.107
>
> 325000
>
> 0
>
> 1d 0h 0m 0s<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945?window=86400>
>
> 651880
>
> 326400
>
> 16.107
>
> 325000
>
> 0
>
> All time<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945?window=%3Aall-time>
>
> 651880
>
> 326400
>
> 16.107
>
> 325000
>
> 0
> Spouts (All time)
>
> *Id*
>
> *Executors*
>
> *Tasks*
>
> *Emitted*
>
> *Transferred*
>
> *Complete latency (ms)*
>
> *Acked*
>
> *Failed*
>
> *Last error*
>
> ReadEventsSpout<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945/component/ReadEventsSpout>
>
> 1
>
> 1
>
> 326400
>
> 326400
>
> 16.107
>
> 325000
>
> 0
> Bolts (All time)
>
> *Id*
>
> *Executors*
>
> *Tasks*
>
> *Emitted*
>
> *Transferred*
>
> *Capacity (last 10m)*
>
> *Execute latency (ms)*
>
> *Executed*
>
> *Process latency (ms)*
>
> *Acked*
>
> *Failed*
>
> *Last error*
>
> ProcessIncidentTagBolt<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945/component/ProcessIncidentTagBolt>
>
> 20
>
> 20
>
> 325480
>
> 0
>
> 0.000
>
> 17.115
>
> 324980
>
> 8.777
>
> 325000
>
> 0
> Topology Configuration
>
> *Key*
>
> *Value*
>
> dev.zookeeper.path
>
> /tmp/dev-storm-zookeeper
>
> drpc.childopts
>
> -Xmx768m
>
> drpc.invocations.port
>
> 3773
>
> drpc.port
>
> 3772
>
> drpc.queue.size
>
> 128
>
> drpc.request.timeout.secs
>
> 600
>
> drpc.worker.threads
>
> 64
>
> java.library.path
>
> /usr/local/lib:/opt/local/lib:/usr/lib
>
> logviewer.childopts
>
> -Xmx128m
>
> logviewer.port
>
> 8000
>
> nimbus.childopts
>
> -Xmx1024m -Djava.net.preferIPv4Stack=true
>
> nimbus.cleanup.inbox.freq.secs
>
> 600
>
> nimbus.file.copy.expiration.secs
>
> 600
>
> nimbus.host
>
> 10.211.64.227
>
> nimbus.inbox.jar.expiration.secs
>
> 3600
>
> nimbus.monitor.freq.secs
>
> 10
>
> nimbus.reassign
>
> true
>
> nimbus.supervisor.timeout.secs
>
> 60
>
> nimbus.task.launch.secs
>
> 120
>
> nimbus.task.timeout.secs
>
> 30
>
> nimbus.thrift.port
>
> 6627
>
> nimbus.topology.validator
>
> backtype.storm.nimbus.DefaultTopologyValidator
>
> storm.cluster.mode
>
> distributed
>
> storm.id
>
> Incidentstopology-1-1386292945
>
> storm.local.dir
>
> /app/storm
>
> storm.local.hostname
>
> bishnu-centos
>
> storm.local.mode.zmq
>
> false
>
> storm.messaging.transport
>
> backtype.storm.messaging.zmq
>
> storm.thrift.transport
>
> backtype.storm.security.auth.SimpleTransportPlugin
>
> storm.zookeeper.connection.timeout
>
> 15000
>
> storm.zookeeper.port
>
> 2181
>
> storm.zookeeper.retry.interval
>
> 1000
>
> storm.zookeeper.retry.intervalceiling.millis
>
> 30000
>
> storm.zookeeper.retry.times
>
> 5
>
> storm.zookeeper.root
>
> /storm
>
> storm.zookeeper.servers
>
> ["10.211.64.227"]
>
> storm.zookeeper.session.timeout
>
> 20000
>
> supervisor.childopts
>
> -Djava.net.preferIPv4Stack=true
>
> supervisor.enable
>
> true
>
> supervisor.heartbeat.frequency.secs
>
> 5
>
> supervisor.monitor.frequency.secs
>
> 3
>
> supervisor.slots.ports
>
> [6700 6701 6702 6703]
>
> supervisor.worker.start.timeout.secs
>
> 120
>
> supervisor.worker.timeout.secs
>
> 30
>
> task.heartbeat.frequency.secs
>
> 3
>
> task.refresh.poll.secs
>
> 10
>
> topology.acker.executors
>
> 2
>
> topology.builtin.metrics.bucket.size.secs
>
> 60
>
> topology.debug
>
> false
>
> topology.disruptor.wait.strategy
>
> com.lmax.disruptor.BlockingWaitStrategy
>
> topology.enable.message.timeouts
>
> true
>
> topology.error.throttle.interval.secs
>
> 10
>
> topology.executor.receive.buffer.size
>
> 1024
>
> topology.executor.send.buffer.size
>
> 1024
>
> topology.fall.back.on.java.serialization
>
> true
>
> topology.kryo.decorators
>
> []
>
> topology.kryo.factory
>
> backtype.storm.serialization.DefaultKryoFactory
>
> topology.kryo.register
>
> topology.max.error.report.per.interval
>
> 5
>
> topology.max.spout.pending
>
> 20000
>
> topology.max.task.parallelism
>
> topology.message.timeout.secs
>
> 90
>
> topology.name
>
> Incidentstopology
>
> topology.optimize
>
> true
>
> topology.receiver.buffer.size
>
> 8
>
> topology.skip.missing.kryo.registrations
>
> false
>
> topology.sleep.spout.wait.strategy.time.ms
>
> 1
>
> topology.spout.wait.strategy
>
> backtype.storm.spout.SleepSpoutWaitStrategy
>
> topology.state.synchronization.timeout.secs
>
> 60
>
> topology.stats.sample.rate
>
> 0.05
>
> topology.tasks
>
> topology.tick.tuple.freq.secs
>
> topology.transfer.buffer.size
>
> 1024
>
> topology.trident.batch.emit.interval.millis
>
> 500
>
> topology.tuple.serializer
>
> backtype.storm.serialization.types.ListDelegateSerializer
>
> topology.worker.childopts
>
> topology.worker.shared.thread.pool.size
>
> 4
>
> topology.workers
>
> 4
>
> transactional.zookeeper.port
>
> transactional.zookeeper.root
>
> /transactional
>
> transactional.zookeeper.servers
>
> ui.childopts
>
> -Xmx768m -Djava.net.preferIPv4Stack=true
>
> ui.port
>
> 8080
>
> worker.childopts
>
> -Xmx3072m -Djava.net.preferIPv4Stack=true
>
> worker.heartbeat.frequency.secs
>
> 1
>
> zmq.hwm
>
> 0
>
> zmq.linger.millis
>
> 5000
>
> zmq.threads
>
> 1
>
>
>
RE: Storm Scalability : Topology to process 5 million records
Posted by Bishnu Patro <Bi...@symantec.com>.
Hi,
I am using RabbitMQ as my event source for my spout code.
I have created a batch 1000 records as 1 message and push it into rabbitmq.
So, for 5 million records, I have pushed 5000 messages (5000000/1000) to RabbitMQ. Each message size is 309 KB.
The spout code reads this message and parse it and emits 1000 tuples to bolt.
Thanks
Bishnu
From: Svend Vanderveken [mailto:svend.vanderveken@gmail.com]
Sent: 06 December 2013 16:45
To: user@storm.incubator.apache.org
Subject: Re: Storm Scalability : Topology to process 5 million records
HI,
Side note for our non-indian friends: a "lac" is 100000 :D
Bishnu, it seems indeed your topology has read and processed 325000 storm tuples, do you confirm the topology is using one storm tuple for each of the 5 million records you mention, or are they regrouped somehow?
Also, note that Storm is usually designed to process "infinite" streams of data, there's not really a concept of a start and an end of a stream. Especially, the error-handling mechanism based on replay of tuples (see doc on Opaque/transactional) imply that some tuples could be processed several times, thus increasing the counts you see in the stats.
If you want to know when a certain number of different tuples have been processed, the only way I can imagine is actually counting yourself the number of distinct record ids you send to RabbitMQ (and make sure you do so in a scalable manner).
Cheers
S
On Fri, Dec 6, 2013 at 11:56 AM, Bishnu Patro <Bi...@symantec.com>> wrote:
Hi,
I have executed the topology with following configurations for 5 million records:
Topology configurations:
Number of spout: 1, Number of executor: 1
Number of Bolt: 1, Number of executor: 20
Number of workers: 4
Number of acks: 2
Number of spout pending: 20000
Optimize flag: true
Debug flag: false
My topology is running on a single machine with zookeeper/nimbus/supervisor as different process.
I am using Rabbit MQ for pushing messages which will be consumed by spout and output of bolt is pushed back to Rabbit MQ in a different queue.
Machine Configurations
RAM: 32 GB
CPU: 8 core
HDD: 500 GB
After some time ( > 3hrs ) when I saw the topology summary (psb) in the storm UI, I don't see any error, but number of emitted is showing only close to 6.5 lacs and number of acked is showing 3.25 lacs, none of messages has failed, although all 5 millions tuples are pushed to RabbitMQ.
Questions:
What is the reason for not showing 5 million messages and how can I conclude that all my messages are processed ?
How to read the topology summary, If I want to find the time taken/throughput to execute 5 million tuples ?
Please forgive for this long mail.
Topology summary
Name
Id
Status
Uptime
Num workers
Num executors
Num tasks
Incidentstopology
Incidentstopology-1-1386292945
ACTIVE
3h 15m 44s
4
23
23
Topology actions
Topology stats
Window
Emitted
Transferred
Complete latency (ms)
Acked
Failed
10m 0s<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945?window=600>
0
0
0.000
0
0
3h 0m 0s<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945?window=10800>
651880
326400
16.107
325000
0
1d 0h 0m 0s<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945?window=86400>
651880
326400
16.107
325000
0
All time<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945?window=%3Aall-time>
651880
326400
16.107
325000
0
Spouts (All time)
Id
Executors
Tasks
Emitted
Transferred
Complete latency (ms)
Acked
Failed
Last error
ReadEventsSpout<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945/component/ReadEventsSpout>
1
1
326400
326400
16.107
325000
0
Bolts (All time)
Id
Executors
Tasks
Emitted
Transferred
Capacity (last 10m)
Execute latency (ms)
Executed
Process latency (ms)
Acked
Failed
Last error
ProcessIncidentTagBolt<http://10.211.64.227:8080/topology/Incidentstopology-1-1386292945/component/ProcessIncidentTagBolt>
20
20
325480
0
0.000
17.115
324980
8.777
325000
0
Topology Configuration
Key
Value
dev.zookeeper.path
/tmp/dev-storm-zookeeper
drpc.childopts
-Xmx768m
drpc.invocations.port
3773
drpc.port
3772
drpc.queue.size
128
drpc.request.timeout.secs
600
drpc.worker.threads
64
java.library.path
/usr/local/lib:/opt/local/lib:/usr/lib
logviewer.childopts
-Xmx128m
logviewer.port
8000
nimbus.childopts
-Xmx1024m -Djava.net.preferIPv4Stack=true
nimbus.cleanup.inbox.freq.secs
600
nimbus.file.copy.expiration.secs
600
nimbus.host
10.211.64.227
nimbus.inbox.jar.expiration.secs
3600
nimbus.monitor.freq.secs
10
nimbus.reassign
true
nimbus.supervisor.timeout.secs
60
nimbus.task.launch.secs
120
nimbus.task.timeout.secs
30
nimbus.thrift.port
6627
nimbus.topology.validator
backtype.storm.nimbus.DefaultTopologyValidator
storm.cluster.mode
distributed
storm.id<http://storm.id>
Incidentstopology-1-1386292945
storm.local.dir
/app/storm
storm.local.hostname
bishnu-centos
storm.local.mode.zmq
false
storm.messaging.transport
backtype.storm.messaging.zmq
storm.thrift.transport
backtype.storm.security.auth.SimpleTransportPlugin
storm.zookeeper.connection.timeout
15000
storm.zookeeper.port
2181
storm.zookeeper.retry.interval
1000
storm.zookeeper.retry.intervalceiling.millis
30000
storm.zookeeper.retry.times
5
storm.zookeeper.root
/storm
storm.zookeeper.servers
["10.211.64.227"]
storm.zookeeper.session.timeout
20000
supervisor.childopts
-Djava.net.preferIPv4Stack=true
supervisor.enable
true
supervisor.heartbeat.frequency.secs
5
supervisor.monitor.frequency.secs
3
supervisor.slots.ports
[6700 6701 6702 6703]
supervisor.worker.start.timeout.secs
120
supervisor.worker.timeout.secs
30
task.heartbeat.frequency.secs
3
task.refresh.poll.secs
10
topology.acker.executors
2
topology.builtin.metrics.bucket.size.secs
60
topology.debug
false
topology.disruptor.wait.strategy
com.lmax.disruptor.BlockingWaitStrategy
topology.enable.message.timeouts
true
topology.error.throttle.interval.secs
10
topology.executor.receive.buffer.size
1024
topology.executor.send.buffer.size
1024
topology.fall.back.on.java.serialization
true
topology.kryo.decorators
[]
topology.kryo.factory
backtype.storm.serialization.DefaultKryoFactory
topology.kryo.register
topology.max.error.report.per.interval
5
topology.max.spout.pending
20000
topology.max.task.parallelism
topology.message.timeout.secs
90
topology.name<http://topology.name>
Incidentstopology
topology.optimize
true
topology.receiver.buffer.size
8
topology.skip.missing.kryo.registrations
false
topology.sleep.spout.wait.strategy.time.ms<http://topology.sleep.spout.wait.strategy.time.ms>
1
topology.spout.wait.strategy
backtype.storm.spout.SleepSpoutWaitStrategy
topology.state.synchronization.timeout.secs
60
topology.stats.sample.rate
0.05
topology.tasks
topology.tick.tuple.freq.secs
topology.transfer.buffer.size
1024
topology.trident.batch.emit.interval.millis
500
topology.tuple.serializer
backtype.storm.serialization.types.ListDelegateSerializer
topology.worker.childopts
topology.worker.shared.thread.pool.size
4
topology.workers
4
transactional.zookeeper.port
transactional.zookeeper.root
/transactional
transactional.zookeeper.servers
ui.childopts
-Xmx768m -Djava.net.preferIPv4Stack=true
ui.port
8080
worker.childopts
-Xmx3072m -Djava.net.preferIPv4Stack=true
worker.heartbeat.frequency.secs
1
zmq.hwm
0
zmq.linger.millis
5000
zmq.threads
1