gmane.comp.java.clojure.storm

Joining of Data

Jon — 2013-05-25T19:42:19

Hi, we are working on trying to convert some of our processing to
Storm, but I am having a few conceptual issues...our workflow is
basically as follows:
(1) a user submits a query
(2) that query is executed, and relevant data is pulled
(3) that data is processed in parallel

(4) more processing of that data is done -- over the entire query's
results


clearly, the sticking point is step (4). Is there a good way to join
all of the results of that query together, to allow for the
processing? The only real things I could think of would be (a)
manually keeping track of each query, and batching it in a bolt or (b)
having two spouts, where steps 1-3 would be in one topology, waiting
for the ack, and then submitting the results to another spout.

Is this the right way to approach this problem?

Excellent performance stories about Storm

Douglas Moore — 2013-05-25T17:16:34

Many companies are now using Storm in production, many more are looking to 
use Storm for stream processing.
It would be great if current users would share stories of performance 
success.

Suggested information to share:

   1. Throughput (messages/sec), 
   2. Storm version, 
   3. # worker nodes, hardware spec, on premise or cloud
   4. Brief description of your use case(s)
   5. Production or benchmark?
   6. Using Trident?
   7. Typical message size
   8. Front end queuing system
   9. Back end persistence.... (e.g. Cassandra, HDFS, Kafka)
   10. Other major architectural components directly related to your Storm 
   Cluster (e.g. Mesos, Memcached, virtualization,...)
   11. How long has Storm been in production?
   12. Company

Thank you in advance!

Storm cannot serialize Google Guava ImmutableList class

preetam joshi — 2013-05-25T00:10:15

Hi,

I have an immutable class that is similar to this one:

    import com.google.common.collect.ImmutableList;

    class SomeObject<X, Y>{
          private final List<X> xList;
          private final List<Y> yList;


          public SomeObject(){
                     this(null, null);
          }

          public SomeObject(List<X> xList, List<Y> yList){
                        this.xList = ImmutableList.copyOf(xList);
                        this.yList = ImmutableList.copyOf(yList);
          }
    }

   class IMPL extends SomeObject<AvroObjectX, AvroObjectY>{
            public IMPL(){
                    super();
            }
           
             public IMPL(List< AvroObjectX > xList, List< AvroObjectY > 
yList){
                    super(xList, yList);
            }      
          
   }

// AvroObjectX and AvroObjectY are AVRO generated classes using their 
respective avro schemas. They *do not implement Serializable *but have 
setters and getters for their fields*.*
*
*
I emit an object of

Storm-Flume connectors contribution

Ravikumar MAV — 2013-05-24T08:56:49

Hello Nathan,

Would like to contribute Flume Storm connectors(Flume Spout and Storm Bolt 
as Avro Sink) to storm-contrib. Currently source code is available < at > 
https://github.com/ExpediaInc/edw-Storm-Flume-Connectors.

Could you give me the access to add the module to storm-contrib? 

Thanks and Regards
Ravi

NullPointerException in StateQueryProcessor.execute - race condition?

Alexander Demidko — 2013-05-24T08:47:43

Hi,

I have a topology consisting of two parts: 
- stream emitting batch of words every 5 seconds and persisting it to the 
accumulator state
- stream emitting tick every 10 seconds and on that tick doing stateQuery 
to that accumulator state

When running it on LocalCluster or remotely using 
StormSubmitter.submitTopology getting same exception:

java.lang.RuntimeException: java.lang.NullPointerException
        at 
backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:87)
        at 
backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:58)
        at 
backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:62)
        at 
backtype.storm.daemon.executor$fn__4050$fn__4059$fn__4106.invoke(executor.clj:658)
        at backtype.storm.util$async_loop$fn__465.invoke(util.clj:377)
        at clojure.lang.AFn.run(AFn.java:24)
        at java.lang.Thread.run(Thread.java:680)
Caused by: java.lang.NullPointerException
        at 
storm.trident.plan

How do i configure maxClientCnxns in zoo.cfg for storm?

Richards Peter — 2013-05-24T10:12:53

Hi,

Today we found that some of the workers on our machines were not getting 
initialised properly. We debugged the issue and found that it was because 
of the default value of maxClientCnxns in zookeeper. The default value is 
10. This means that there cannot be more than 10 connections from a single 
client machine to a single member of zookeeper ensemble. By increasing this 
parameter the issue got resolved. I would like to know how to decide upon 
the value which has to be assigned for this parameter. 

Assume that I have a storm cluster containing m machines each having n 
supervisor ports. I am running nimbus, supervisor and ui on one of these m 
machines. For safer side should I assign the value to be n+3(considering 
nimbus, supervisor and ui on the master machine)?

Thanks,
Richards Peter.

Trident: Querying on a partitioned state.

2013-05-24T08:58:48

Hello,

I have a state partitioned into 3. When I query the state I find that my 
stream used for querying is automatically partitioned into 3. To illustrate 
this point here is the sample code.

TridentState state = topology.newStream("spout",new FixedbatchSpout())
                                          .partitionBy(new 
Fields("some_field"))
                                          .partitionAggregate(new 
Fileds("some_fields..",""...), new TimeSlotBasedAggregator(), new 
Fields("slot","value"))
                                          .partitionPersist(....)
                                          .parallelismHint(3);

topology.newStream("tick_stream",new TickSpout())
            .stateQuery(state, . . .. )
            .parallelismHint(4);

My question is, in how many parts the state will be partitioned?

Regards
Kshitiz

Multiple topologies and worker count

2013-05-23T21:46:21

Hi, in our storm (0.8.2) cluster -- we have 8 supervisor machines and each 
machine by default configuration has 4 slots (supervisor.slots). So our 
total slots are 32. We are deploying two independent topologies and each 
topology has topology.workers set to 8. Does this mean that 16 out of the 
32 slots will be used across the clusters for both the topologies? 

In our environment, I run into the issue where once the first topology is 
submitted (and running fine), the second topology does not start. When I 
explicitly make the topology.workers=6 for 1st topology and 
topology.workers=2 for 2nd topology, they work fine together. I was under 
the impression that # of slots = # of JVMs = # of total workers in the 
cluster? Is that assumption right? If so, I would think that both 
topologies should be able to start independently on the same cluster with 
num of workers set to 8 each

Thanks,
Suyog

autogen.sh: error: could not find pkg-config. pkg-config is required to run autogen.sh.

SAG_Developer — 2013-05-23T16:59:57

I am trying to setup a storm cluster on Ubuntu 12.04 server by following 
this 
link https://github.com/nathanmarz/storm/wiki/Setting-up-a-Storm-cluster

I have ZeroMQ installed
Running autogen.sh in the jzmq source folder is throwing an error.* autogen.sh: 
error: could not find pkg-config.  pkg-config is required to run autogen.sh.
*
*
*
*What do I need on my server to by pass this issue?*

Combining the output of two Trident streams

2013-05-23T20:55:45

If I have two Trident streams that produce the same fields as output, but 
at different rates, is there a way to combine their output?

So S1 has fields X, Y and Z and tuples entering it: 
x1, y1, z1
x2, y2, z2
x3, y3, z3
.
.
.

And S2 has the same fields X, Y and Z, with tuples:
x10, y10, z10
x11, y11, z11
.
.

Can I combine them to a new stream containing:

x1, y1, z1
x2, y2, z2
x3, y3, z3
x10, y10, z10
x11, y11, z11
.
.
.

I can see there's the TridentTopology.multireduce method, but I'm not sure 
if that'll do the job, or what the implications are for the flow of tuples 
from
each input stream?

Ack/Fail counting when topology fails

phlp — 2013-05-23T17:21:31

We use BaseRichSpout and BaseRichBolt in our application.  

The spout counts emitted tuples and acks/fails to derive 
state information when emitted-count == sum(ack-count, fail-count).

When a tuple is emitted in one topology which is then restarted before the 
tuple is ack'd by the bolt, will the new spout's ack routine be called when 
the tuple is re-emitted by storm and processed and ack'd by the bolts?  

I.E. will a spout get an ack for a tuple that it did not directly emit?

I'm assuming the answer is yes so that (for instance) external state 
associated with the tuple is cleaned up.

trident batch question

Sunil Yarram — 2013-05-23T01:22:19

Hi,
  I have few basic questions on trident processing:

1. The trident document mentions only one batch will be in commit phase, 
does that mean there can be many parallel batches doing processing but not 
doing the commit?
2. If trident is processing one batch at a time then the parallelism is 
achevied by dividing the batch into partitions?  and i am guessing the 
number of partitions are controlled by parallelismHint() method.

Thanks,
Sunil.

About setting RotatingMap ExpireTime

2013-05-22T07:57:56

Hi, all fella:
   I recently utilize storm for realtime computing. I required a 
TimePeriodMemoryCache, then I found TimeCacheMap that is annotated 
 Deprecated. I found this link 
http://comments.gmane.org/gmane.comp.java.clojure.storm/4683 to recommand 
RotatingMap instead of TimeCacheMap. I seek its api 
http://nathanmarz.github.io/storm/doc/backtype/storm/utils/RotatingMap.html, 
however, I NOT where to set expiredTime.

Please help me.

Best Begard.

Robin.

storm-config, topologyLoader

Stumpyfr . — 2013-05-22T19:32:10

Hi,

I didn't find something like that, so I created it.
Just a very first version but I wanted to share it with you and maybe get 
some advices from the community.

The main reason was to not need to compile my .jar each time I want try a 
new configuration for the parallelism of my topology.

You still need to kill/start your topology but to change the configuration, 
just need to change the values inside the json configuration file.
I plan to add some configuration values, open to any ideas to what will be 
the more useful.

./storm-0.8.2/bin/storm jar MyProject.jar com.tekforge.StormSample.App 
config.json

https://github.com/stumpyfr/storm-config

(Newbie) Is there a very simple example to get started?

Miguel Durazo — 2013-05-22T20:03:55

I already read the wiki, but i have no idea how to begin, i want to be able 
to test simple things locally first, a small code example would do

Thanks!

Trident ParallelismHint Set But Only One Work

2013-05-22T15:10:38

Hi, all
  I am using Trident topology and I am confused about the parallelism, I 
found 
https://groups.google.com/forum/?fromgroups#!topic/storm-user/QlCmDmnEghs and < at >nathanmarz 
explained what the parallelism use, I build my topology below:

* List<HostPort> hostports = new ArrayList<HostPort>();*
* HostPort hostport = new HostPort(getPropertyVar("KAFKA_DOMAIN1"),*
* Integer.parseInt(getPropertyVar("KAFKA_PORT")));*
* hostports.add(hostport);*
* hostport = new HostPort(getPropertyVar("KAFKA_DOMAIN2"),*
* Integer.parseInt(getPropertyVar("KAFKA_PORT")));*
* hostports.add(hostport);*
* TridentKafkaConfig.StaticHosts hosts = new TridentKafkaConfig.StaticHosts(
*
* Lists.newArrayList(hostport), 1);*
*
*
* TridentKafkaConfig config = new TridentKafkaConfig(hosts,*
* getPropertyVar("KAFKA_TOPIC"));*
* // config.scheme = new SchemeAsMultiScheme(new StringScheme());*
* // config.forceFromStart = true;*
* TransactionalTridentKafkaSpout spout = new TransactionalTridentKafkaSpout(
*
* config);*
*

Trident ParallelismHint Set But Only One Work

2013-05-22T15:06:24

Hi, all
  I am using Trident topology and I am confused about the parallelism, I 
found 
https://groups.google.com/forum/?fromgroups#!topic/storm-user/QlCmDmnEghs and < at >nathanmarz 
explained what the parallelism use, I build my topology below:

* List<HostPort> hostports = new ArrayList<HostPort>();*
* HostPort hostport = new HostPort(getPropertyVar("KAFKA_DOMAIN1"),*
* Integer.parseInt(getPropertyVar("KAFKA_PORT")));*
* hostports.add(hostport);*
* hostport = new HostPort(getPropertyVar("KAFKA_DOMAIN2"),*
* Integer.parseInt(getPropertyVar("KAFKA_PORT")));*
* hostports.add(hostport);*
* TridentKafkaConfig.StaticHosts hosts = new TridentKafkaConfig.StaticHosts(
*
* Lists.newArrayList(hostport), 1);*
*
*
* TridentKafkaConfig config = new TridentKafkaConfig(hosts,*
* getPropertyVar("KAFKA_TOPIC"));*
* // config.scheme = new SchemeAsMultiScheme(new StringScheme());*
* // config.forceFromStart = true;*
* TransactionalTridentKafkaSpout spout = new TransactionalTridentKafkaSpout(
*
* config);*
*

spout stopped send message, blocked in SpoutOutputCollector.emit()

陈竞 — 2013-05-22T08:15:41

my topology has 24spout,  24bolt, and 24workers,  every worker contains a
spout and a bolt,  after running a few hours, some spouts would be blocked,
do not send any message,  i use jstack to check thread stack, i get the
following result:


"Thread-15" prio=10 tid=0x00002aabb86c1c00 nid=0xc53 waiting on condition
[0x00000000454de000..0x00000000454dea10]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00002aaab483d8a8> (a
java.util.concurrent.locks.ReentrantLock$NonfairSync)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:158)
        at
java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:747)
        at
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:778)
        at
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1114)
        at
java.util.concurrent.locks

Aggregating all of the Storm Logs

Gary Malouf — 2013-05-21T21:08:58

With the large quantity of log files produced by storm by default, I was 
wondering how others were rolling up the logs into something that can be 
easily monitored for ERROR messages.

0.8.2 setup

Michael Black — 2013-05-21T20:19:54

I’m trying to set up a 3-node system on 0.8.2.
I want one node to be ui, nimbus and a supervisor (vmddcdmaster)
I want the others to be just supervisors (vmddcdnode1, vmddcdnode2)
I’m going to write a customer scheduler as my workload is pretty well known 
so node1&node2 will be running jobs.
The master will just be the top node which coordinates things.

So..I have this same storm.yaml on all 3 nodes.  I run nimbus, supervisor, 
and ui on the master and just supervisor on the 2 nodes.

Master:
supervisor.slots.ports:
     - 6700
     - 6701
     - 6702
########### These MUST be filled in for a storm configuration
storm.zookeeper.servers:
     - "vmddcdmaster"
     - "vmddcdnode1"
     - "vmddcdnode2"
#
storm.local.dir: "/var/storm"
nimbus.host: "vmddcdmaster"

The UI shows only node1 in the Supervisor summary – master showed up for a 
short while then disappeared.  I expected all 3 machines to show up.
I imagine I’m doing something wrong but it’s not clear to me what is wrong 
and there are no err

python DRPC client

Homer Strong — 2013-05-21T19:07:08

I've noticed that

https://github.com/nathanmarz/storm/tree/master/storm-core/src/py

seems to contains a python DRPC client, looks like generated from thrift, 
but I haven't seen any docs on it or heard of folks using it.

Has anybody had any experience using it?