Emitting Custom Java Object as Tuple from Spout

I am new to Apache Storm. To understand it I was looking at Storm examples provided in the Storm tutorial (https://github.com/apache/storm/tree/master/examples/storm-starter) .

In all the examples, we are emitting primitive types (String,int etc) as tuples.

In my use case I am having a Java object which I want to emit as tuple to bolts.

I am not sure how to do it. It seems I have to implement a custom Tuple producer to convert Java object to Values.

Can anyone provide me some example how to do that:

For ex my Java object is :

class Bean 
{
   int A;
   String B;
   Bean2 b;
   //setters and getters
}

and

class Bean2
{
    //Some Attributes
}

Now, in my nextTuple() method for Spout, I have an instance of a Bean object.

class Spout implements IRichSpout
{
    //...
    void nextTuple()
    {
        Bean b = queue.poll();//queue contains instances of Bean Object
        //How to convert this object to tuple and emit it as Tuples???
    }
}

How do I translate into Tuple and emit it and consume it through my bolt.

Solution

There are two possibilities:

Emit just a single attribute tuple that contains the Bean:
```
class Spout implements IRichSpout {
    void nextTuple() {
        Bean b = queue.poll();
        collector.emit(new Values(b));
    }
}
```
This is the simplest approach, but has the disadvantage, that you cannot use the individual attributes in fieldGrouping(...) for consuming bolts. Furthermore, you should register Bean in Kryo for efficient (de)serialization:
```
Config c = new Config();
c.registerSerialization(Bean.class);
StormSubmitter.submitTopology("name", c, builder.createTopology());
```
Emit each attribute individually:
```
class Spout implements IRichSpout {
    void nextTuple() {
        Bean b = queue.poll();
        // of course, you can also un-nest Bean2
        collector.emit(new Values(b.getA(), b.getB(), b.getBean2()));
    }
}
```
This allows to use each attribute (or combination of attributes) in fieldsGrouping(...). However, at the consumer, you need to retrieve each attribute individually and use your setter-methods to build a new Bean object.

As an alternative to the second approach, please see How to use apache storm tuple