Storm & Esper

At work, we recently started using Esper for realtime analytics, and so far we quite like Esper. It is a great tool at what it does – running queries continuously over data. The problem however then becomes how to get data into Esper. The recently released Storm could be one way to do that, so I got curios and started playing around with it to see if it could be made to work with Esper. And it turns out, the integration is straightforward.

Some Storm basics

Storm has three basic concepts that are relevant in this context: streams, spouts, and bolts. At the core, Storm facilitates data transfer between spouts and bolts using streams of tuples.

Spouts are the basic data emitters, typically retrieving the data from outside of the Storm cluster. A simple example of this would be a spout that retrieves the tweet stream via the Twitter API and emits the tweets as a stream into the Storm cluster.

Bolts are data processors that receive one or more streams as input and potentially also emit (processed) data on one or more streams. In the twitter example, one could for instance imagine bolts that count the number of tweets per second, or detect the language of the tweet and reemit the tweets into per-language streams.

The data in the streams has a simple tuple form consisting of a fixed number of named values called fields. Storm does not care about the data types of the individual fields in the tuple as long as they can be serialized to the wire format (which is Thrift), whether via serializers provided by Storm or custom ones. Spouts and bolts need to declare the number of fields and their names for each of the tuples they are going to emit as part of the initial setup of the topology. This also means that the number of fields and their names are fixed for the duration of a Storm ‘session’.

Some Eper basics

Esper is, and I’m simplifying things quite a bit here, a processing engine for data streams that uses queries run on the data streams to processes them. Think of it as a way to run SQL-like queries on data that streams by. The queries run continuously and thus have a time or amount-of-data aspect to them. Continuing the twitter example from above, if we consider the never-ending stream of tweets as the data stream that Esper works with, then an Esper query could for instance return the number of tweets per second like so:

select count(*) as tps from Twitter.win:time_batch(1 sec)

The time_batch part in this example will direct Esper to apply the count function on 1-sec batches of events.

Esper data streams consist of structured data called events. The types of these events can be POJOs, maps, and other things. Events are typically registered with Esper in advance before submitting a query to Esper. This means that you have to tell Esper about which kind of event type you give it (java class, map, …) and which properties the event type has. For java classes, Esper can figure that out itself but for map events you need to tell Esper explicitly about the possible keys and the value data types. Fortunately, Esper is forgiving when it comes to the data types. You can tell it that you’ll give it Objects, and it will happily accept numbers in the actual data stream and perform numeric operations on them.

How to combine the two

Storm’s tuples are quite similar to Esper’s map event types. The tuple field names map naturally to map keys and the field values to values for these keys. The tuple fields are not typed when they are defined, but that does not pose a big problem for us as we can simply tell Esper that they are of type Object. In addition, the fact that tuples have to be defined before a topology is run, makes it relatively easy for us to define the map event type in the setup phase.

I am going to use the twitter stream example from the storm-starter project to show how you can use Esper to count the number of tweets per second and also find the maximum number of retweets per 1 second interval. This is probably not of great practical use, but will show off some aspects of the Storm – Esper integration.

An up-to-date version of this code is available on GitHub.

Let’s get started with the twitter spout, a slightly adapted version of the one from the storm-starter project:

public class TwitterSpout implements IRichSpout, StatusListener {
    private static final long serialVersionUID = 1L;

    private final String username;
    private final String pwd;
    private transient BlockingQueue<Status> queue;
    private transient SpoutOutputCollector collector;
    private transient TwitterStream twitterStream;

    public TwitterSpout(String username, String pwd) {
        this.username = username;
        this.pwd = pwd;
    }

    @Override
    public boolean isDistributed() {
        return false;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("createdAt", "retweetCount"));
    }

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.queue = new ArrayBlockingQueue<Status>(1000);
        this.collector = collector;

        Configuration twitterConf = new ConfigurationBuilder().setUser(username)
                                                              .setPassword(pwd)
                                                              .build();
        TwitterStreamFactory fact = new TwitterStreamFactory(twitterConf);

        twitterStream = fact.getInstance();
        twitterStream.addListener(this);
        twitterStream.sample();
    }

    @Override
    public void onStatus(Status status) {
        queue.offer(status);
    }

    @Override
    public void nextTuple() {
        Status value = queue.poll();
        if (value == null) {
            Utils.sleep(50);
        }
        else {
            collector.emit(tuple(value.getCreatedAt().getTime(),
                                 value.getRetweetCount()));
        }
    }

    @Override
    public void close() {
        twitterStream.shutdown();
    }

    @Override
    public void ack(Object arg0) {}
    @Override
    public void fail(Object arg0) {}
    @Override
    public void onException(Exception ex) {}
    @Override
    public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {}
    @Override
    public void onTrackLimitationNotice(int numberOfLimitedStatuses) {}
    @Override
    public void onScrubGeo(long userId, long upToStatusId) {}
}

This defines a spout that emits a single stream of tuples with two fields, createdAt (a timestamp) and retweetCount (an integer).

You’ll notice that aside from the twitter username and password, all fields in the spout are marked as transient, and initialized in the open method. The reason for this is that Storm requires spouts and bolts to be serializable so it can move them to some node in the Storm cluster before starting the topology.

The Esper bolt itself is generic. You pass it esper statements and the names of the output fields which will be generated by these esper statements. The adapted main method for our Twitter example looks like this:

public static void main(String[] args) {
    final String username = args[0];
    final String pwd = args[1];

    TopologyBuilder builder = new TopologyBuilder();
    TwitterSpout spout = new TwitterSpout(username, pwd);
    EsperBolt bolt = new EsperBolt(
        new Fields("tps", "maxRetweets"),
        "select count(*) as tps, max(retweetCount) as maxRetweets from Storm.win:time_batch(1 sec)");

    builder.setSpout(1, spout);
    builder.setBolt(2, bolt).shuffleGrouping(1);

    Config conf = new Config();
    conf.setDebug(true);

    LocalCluster cluster = new LocalCluster();

    cluster.submitTopology("test", conf, builder.createTopology());
    Utils.sleep(10000);
    cluster.shutdown();
}

Note how the Esper statement returns tps and maxRetweets which are also declared as the two output fields for the bolt.

The bolt code itself (see a version of this that is kept up-to-date with Storm here) consists of three pieces. The setup part constructs map event types for each input stream and registers them with Esper (I omitted the Esper setup code):

private void setupEventTypes(TopologyContext context, Configuration configuration) {
    Set<GlobalStreamId> sourceIds = context.getThisSources().keySet();
    singleEventType = (sourceIds.size() == 1);

    for (GlobalStreamId id : sourceIds) {
        Map<String, Object> props = new LinkedHashMap<String, Object>();

        setupEventTypeProperties(
            context.getComponentOutputFields(id.get_componentId(),
                                             id.get_streamId()),
                                             props);
        configuration.addEventType(getEventTypeName(id.get_componentId(),
                                                    id.get_streamId()),
                                                    props);
    }
}

private String getEventTypeName(int componentId, int streamId) {
    if (singleEventType) {
        return "Storm";
    }
    else {
        return String.format("Storm_%d_%d", componentId, streamId);
    }
}

private void setupEventTypeProperties(Fields fields, Map<String, Object> properties){
    int numFields = fields.size();

    for (int idx = 0; idx < numFields; idx++) {
        properties.put(fields.get(idx), Object.class);
    }
}

The field-to-property mapping is straightforward. It simply registers properties of type Object using the field names in the event type corresponding to the input stream. If the bolt only has a single input stream, then it registers a single event type called Storm. For multiple types, it uses the component id (id of the spout or bolt that the data comes from) and the stream id (spouts and bolts can emit multiple streams) to generate a name Storm_{component id}_{stream id}.

The second part is the transfer of data from Storm to Esper:

@Override
public void execute(Tuple tuple) {
    String eventType = getEventTypeName(tuple.getSourceComponent(),
                                        tuple.getSourceStreamId());
    Map<String, Object> data = new HashMap<String, Object>();
    Fields fields = tuple.getFields();
    int numFields = fields.size();

    for (int idx = 0; idx < numFields; idx++) {
        String name = fields.get(idx);
        Object value = tuple.getValue(idx);

        data.put(name, value);
    }

    runtime.sendEvent(data, eventType);
}

This method is called by Storm whenever a tuple from any of the connected streams is sent to the bolt. The code therefore first has to find the event type name corresponding to the tuple. Then it iterates over the fields in the tuple and puts the values into a map using the field names as the keys. Finally, it passes that map to Esper.

At this moment, Esper will route this map (the event) through the statements which in turn might produce new data that we need to hand back to Storm. For this purpose, the bolt registered itself as a listener for data emitted from any of the statements that we configured during the setup. Esper will then call back the update method on the bolt if one of the statements generated data. The update method will then basically perform the reverse operation of the execute method and convert the event data to a tuple:

@Override
public void update(EventBean[] newEvents, EventBean[] oldEvents) {
    if (newEvents != null) {
        for (EventBean newEvent : newEvents) {
            collector.emit(toTuple(newEvent));
        }
    }
}

private List<Object> toTuple(EventBean event) {
    int numFields = outputFields.size();
    List<Object> tuple = new ArrayList<Object>(numFields);

    for (int idx = 0; idx < numFields; idx++) {
        tuple.add(event.get(outputFields.get(idx)));
    }
    return tuple;
}