PubNub Spout for Storm - Grainier Perera

What is PubNub?

PubNub is a real-time messaging platform which uses publisher / subscriber pattern. You can find out more about PubNub by visiting www.pubnub.com

What is Storm?

Storm is a free and open source distributed realtime computation system. Storm can be used in scenarios like: realtime analytics, online machine learning, continuous computation, distributed RPC, ETL, and more. To learn more about Storm project, visit Storm Project website

PubNub Spout for Storm

Once you get familiar with Storm, you would know that Storm uses a concept called Spouts. A Spout is a source of streams in a Storm topology. Basically Spout is the one who will read tuples from the external source and emit them in to the Storm topology. Please refer to Storm Wiki to read more about the concept of Spout.

In this post it will show how to use PubNub as the real-time external data source for a Storm topology, by making a PubNub Spout.

The Design

Basically what we do is create a Storm Spout, which is subscribed to a PubNub channel. Once, some data is been published to that channel, Spout will receive those data and it will emit them in to the Storm topology.

The Code

package com.kohls.trending.spout;

import backtype.storm.Config;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;
import com.kohls.trending.TrendingProperties;
import com.pubnub.api.Callback;
import com.pubnub.api.Pubnub;
import com.pubnub.api.PubnubError;
import com.pubnub.api.PubnubException;

import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * This is the PUBNUB Spout. This works as the input data stream for this Topology
 */
@SuppressWarnings({"rawtypes", "serial"})
public class PubnubSpout extends BaseRichSpout {

    Pubnub _pubnub;
    private SpoutOutputCollector collector;
    private LinkedBlockingQueue<String> queue;

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        _pubnub = new Pubnub(TrendingProperties.getInstance().PUBNUB_PUB_KEY, TrendingProperties.getInstance().PUBNUB_SUB_KEY, false);

        queue = new LinkedBlockingQueue<String>(1000);
        this.collector = collector;

        try {
            _pubnub.subscribe(new String[]{TrendingProperties.getInstance().PUBNUB_SUB_CHANNEL}, new Callback() {
                @Override
                public void successCallback(String channel, Object message) {
                    queue.offer(message.toString());
                }

                @Override
                public void errorCallback(String channel, PubnubError error) {
                }
            });
        } catch (PubnubException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void nextTuple() {
        String ret = queue.poll();
        if (ret == null) {
            Utils.sleep(50);
        } else {
            collector.emit(new Values(ret));
        }
    }

    @Override
    public void close() {
        _pubnub.unsubscribe(TrendingProperties.getInstance().PUBNUB_SUB_CHANNEL);
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        Config ret = new Config();
        ret.setMaxTaskParallelism(1);
        return ret;
    }

    @Override
    public void ack(Object id) {
    }

    @Override
    public void fail(Object id) {
    }

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

package com.kohls.trending.spout;

import backtype.storm.Config;

import backtype.storm.spout.SpoutOutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.base.BaseRichSpout;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Values;

import backtype.storm.utils.Utils;

import com.kohls.trending.TrendingProperties;

import com.pubnub.api.Callback;

import com.pubnub.api.Pubnub;

import com.pubnub.api.PubnubError;

import com.pubnub.api.PubnubException;

import java.util.Map;

import java.util.concurrent.LinkedBlockingQueue;

/**

* This is the PUBNUB Spout. This works as the input data stream for this Topology

@SuppressWarnings({"rawtypes", "serial"})

public class PubnubSpout extends BaseRichSpout {

Pubnub _pubnub;

private SpoutOutputCollector collector;

private LinkedBlockingQueue<String> queue;

@Override

public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {

_pubnub = new Pubnub(TrendingProperties.getInstance().PUBNUB_PUB_KEY, TrendingProperties.getInstance().PUBNUB_SUB_KEY, false);

queue = new LinkedBlockingQueue<String>(1000);

this.collector = collector;

try {

_pubnub.subscribe(new String[]{TrendingProperties.getInstance().PUBNUB_SUB_CHANNEL}, new Callback() {

@Override

public void successCallback(String channel, Object message) {

queue.offer(message.toString());

}

@Override

public void errorCallback(String channel, PubnubError error) {

}

});

} catch (PubnubException e) {

e.printStackTrace();

}

@Override

public void nextTuple() {

String ret = queue.poll();

if (ret == null) {

Utils.sleep(50);

} else {

collector.emit(new Values(ret));

}

@Override

public void close() {

_pubnub.unsubscribe(TrendingProperties.getInstance().PUBNUB_SUB_CHANNEL);

}

@Override

public Map<String, Object> getComponentConfiguration() {

Config ret = new Config();

ret.setMaxTaskParallelism(1);

return ret;

}

@Override

public void ack(Object id) {

}

@Override

public void fail(Object id) {

}

@Override

public void declareOutputFields(OutputFieldsDeclarer declarer) {

declarer.declare(new Fields("json"));

}

If you find this useful, Please leave a comment below.