The spout in a transactional topology is completely different from a standard spout.

public class TweetsTransactionalSpout extends BaseTransactionalSpout<TransactionMetadata> {

As you can see in the class definition, TweetsTransactionalSpout extends BaseTransactionalSpout with a generic type. The type you set there is something known as the transaction metadata. It will be used later to emit batches of tuples from the source.

In this example, TransactionMetadata is defined as:

public class TransactionMetadata implements Serializable {
  private static final long serialVersionUID = 1L;

  long from;
  int quantity;

  public TransactionMetadata(long from, int quantity) {
    this.from = from;
    this.quantity = quantity;
  }
}

Here you’ll store from and quantity, which will tell you exactly how to generate the batch of tuples.

To finish the implementation of the spout, you need to implement the following three methods:

@Override
public ITransactionalSpout.Coordinator<TransactionMetadata> getCoordinator(
Map conf, TopologyContext context) {
  return new TweetsTransactionalSpoutCoordinator();
}

@Override
public backtype.storm.transactional.ITransactionalSpout.Emitter<TransactionMetadata> getEmitter(
Map conf, TopologyContext context) {
  return new TweetsTransactionalSpoutEmitter();
}

@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
  declarer.declare(new Fields("txid", "tweet_id", "tweet"));
}

In the getCoordinator method, you tell Storm which class will coordinate the generation of batches of tuples. With getEmitter, you tell Storm which class will be responsible for reading batches of tuples from the source and emitting them to a stream in the topology. And finally, as you did before, you need to declare which fields are emitted.

public class RQ {
  public static final String NEXT_READ = "NEXT_READ";
  public static final String NEXT_WRITE = "NEXT_WRITE";

  Jedis jedis;

  public RQ() {
    jedis = new Jedis("localhost");
  }

  public long getAvailableToRead(long current) {
    return getNextWrite() - current;
  }

  public long getNextRead() {
    String sNextRead = jedis.get(NEXT_READ);
    if(sNextRead == null)
      return 1;
    return Long.valueOf(sNextRead);
  }

  public long getNextWrite() {
    return Long.valueOf(jedis.get(NEXT_WRITE));
  }

  public void close() {
    jedis.disconnect();
  }

  public void setNextRead(long nextRead) {
    jedis.set(NEXT_READ, ""+nextRead);
  }

  public List<String> getMessages(long from, int quantity) {
    String[] keys = new String[quantity];

    for (int i = 0; i < quantity; i++)
      keys[i] = ""+(i+from);

    return jedis.mget(keys);
  }
}

public static class TweetsTransactionalSpoutCoordinator implements ITransactionalSpout.Coordinator<TransactionMetadata> {
  TransactionMetadata lastTransactionMetadata;
  RQ rq = new RQ();
  long nextRead = 0;

  public TweetsTransactionalSpoutCoordinator() {
    nextRead = rq.getNextRead();
  }

  @Override
  public TransactionMetadata initializeTransaction(BigInteger txid,
        TransactionMetadata prevMetadata) {
    long quantity = rq.getAvailableToRead(nextRead);
    quantity = quantity > MAX_TRANSACTION_SIZE ? MAX_TRANSACTION_SIZE : quantity;
    TransactionMetadata ret = new TransactionMetadata(nextRead, (int)quantity);

    nextRead += quantity;
    return ret;
  }

  @Override
  public boolean isReady() {
    return rq.getAvailableToRead(nextRead) > 0;
  }

  @Override
  public void close() {
    rq.close();
  }
}

public static class TweetsTransactionalSpoutEmitter implements ITransactionalSpout.Emitter<TransactionMetadata> {

  RQ rq = new RQ();

  public TweetsTransactionalSpoutEmitter() {
  }

  @Override
  public void emitBatch(TransactionAttempt tx,
        TransactionMetadata coordinatorMeta, BatchOutputCollector collector) {
    rq.setNextRead(coordinatorMeta.from+coordinatorMeta.quantity);
    List<String> messages = rq.getMessages(coordinatorMeta.from,
        coordinatorMeta.quantity);

    long tweetId = coordinatorMeta.from;

    for (String message : messages) {
      collector.emit(new Values(tx, ""+tweetId, message));
      tweetId++;
    }
  }

  @Override
  public void cleanupBefore(BigInteger txid) {
  }

  @Override
  public void close() {
    rq.close();
  }
}

First let’s see the standard bolts of this topology:

public class UserSplitterBolt implements IBasicBolt{
  private static final long serialVersionUID = 1L;

  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declareStream("users", new Fields("txid", "tweet_id", "user"));
  }

  @Override
  public Map<String, Object> getComponentConfiguration() {
    return null;
  }

  @Override
  public void prepare(Map stormConf, TopologyContext context) {
  }

  @Override
  public void execute(Tuple input, BasicOutputCollector collector) {
    String tweet = input.getStringByField("tweet");
    String tweetId = input.getStringByField("tweet_id");
    StringTokenizer strTok = new StringTokenizer(tweet, " ");
    TransactionAttempt tx = (TransactionAttempt)input.getValueByField("txid");
    HashSet<String> users = new HashSet<String>();

    while(strTok.hasMoreTokens()) {
      String user = strTok.nextToken();

      // Ensure this is an actual user, and that it's not repeated in the tweet
      if(user.startsWith("@") && !users.contains(user)) {
        collector.emit("users", new Values(tx, tweetId, user));
        users.add(user);
      }
    }
  }

  @Override
  public void cleanup() {

  }
}

As mentioned earlier in this chapter, UserSplitterBolt receives tuples, parses the text of the tweet, and emits words preceded by @, or the Twitter users. HashtagSplitterBolt works in a very similar way.

public class HashtagSplitterBolt implements IBasicBolt{

  private static final long serialVersionUID = 1L;

  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declareStream("hashtags", new Fields("txid", "tweet_id", "hashtag"));
  }

  @Override
  public Map<String, Object> getComponentConfiguration() {
    return null;
  }

  @Override
  public void prepare(Map stormConf, TopologyContext context) {
  }

  @Override
  public void execute(Tuple input, BasicOutputCollector collector) {
    String tweet = input.getStringByField("tweet");
    String tweetId = input.getStringByField("tweet_id");
    StringTokenizer strTok = new StringTokenizer(tweet, " ");
    TransactionAttempt tx = (TransactionAttempt)input.getValueByField("txid");
    HashSet<String> words = new HashSet<String>();

    while(strTok.hasMoreTokens()) {
      String word = strTok.nextToken();

      if(word.startsWith("#") && !words.contains(word)) {
        collector.emit("hashtags", new Values(tx, tweetId, word));
        words.add(word);
      }
    }
  }

  @Override
  public void cleanup() {
  }
}

Now let’s see what happens in UserHashtagJoinBolt. The first important thing to notice is that it is a BaseBatchBolt. This means that the execute method will operate on the received tuples but won’t be emitting any new tuple. Eventually, when the batch is finished, Storm will call the finishBatch method.

public void execute(Tuple tuple) {
  String source = tuple.getSourceStreamId();
  String tweetId = tuple.getStringByField("tweet_id");

  if("hashtags".equals(source)) {
    String hashtag = tuple.getStringByField("hashtag");
    add(tweetHashtags, tweetId, hashtag);
  } else if("users".equals(source)) {
    String user = tuple.getStringByField("user");
    add(userTweets, user, tweetId);
  }
}

Since you need to associate all the hashtags of a tweet with the users mentioned in that tweet and count how many times they appeared, you need to join the two streams of the previous bolts. Do that for the entire batch, and once it finishes, the finishBatch method is called.

@Override
public void finishBatch() {

  for (String user : userTweets.keySet()) {
    Set<String> tweets = getUserTweets(user);
    HashMap<String, Integer> hashtagsCounter = new HashMap<String, Integer>();
    for (String tweet : tweets) {
      Set<String> hashtags = getTweetHashtags(tweet);
      if(hashtags != null) {
        for (String hashtag : hashtags) {
          Integer count = hashtagsCounter.get(hashtag);
          if(count == null)
            count = 0;
          count ++;
          hashtagsCounter.put(hashtag, count);
        }

      }
    }

    for (String hashtag : hashtagsCounter.keySet()) {
      int count = hashtagsCounter.get(hashtag);
      collector.emit(new Values(id, user, hashtag, count));
    }
  }
}

In this method, generate and emit a tuple for each user-hashtag, and the amount of times it occurred.

You can see the complete implementation in the downloadable code available on GitHub.

As you’ve learned, batches of tuples are sent by the coordinator and emitters across the topology. Those batched of tuples are processed in parallel without any specific order.

The coordinator bolts are special batch bolts that implement ICommitter or have been set with setCommiterBolt in the TransactionalTopologyBuilder. The main difference with regular batch bolts is that the finishBatch method of committer bolts executes when the batch is ready to be committed. This happens when all previous transactions have been committed successfully. Additionally, finishBatch method is executed sequentially. So if the batch with transaction ID 1 and the batch with transaction ID 2 are being processed in parallel in the topology, the finishBatch method of the committer bolt that is processing the batch with transaction ID 2 will get executed only when the finishBatch of batch with transaction ID 1 has finished without any errors.

The implementation of this class follows:

public class RedisCommiterCommiterBolt extends BaseTransactionalBolt
    implements ICommitter {
  public static final String LAST_COMMITED_TRANSACTION_FIELD = "LAST_COMMIT";
  TransactionAttempt id;
  BatchOutputCollector collector;
  Jedis jedis;


  @Override
  public void prepare(Map conf, TopologyContext context,
        BatchOutputCollector collector, TransactionAttempt id) {
    this.id = id;
    this.collector = collector;
    this.jedis = new Jedis("localhost");
  }

  HashMap<String, Long> hashtags = new HashMap<String, Long>();
  HashMap<String, Long> users = new HashMap<String, Long>();
  HashMap<String, Long> usersHashtags = new HashMap<String, Long>();

  private void count(HashMap<String, Long> map, String key, int count) {
    Long value = map.get(key);
    if(value == null)
    value = (long) 0;
    value += count;
    map.put(key, value);
  }

  @Override
  public void execute(Tuple tuple) {
    String origin = tuple.getSourceComponent();
    if("users-splitter".equals(origin)) {
      String user = tuple.getStringByField("user");
      count(users, user, 1);
    } else if("hashtag-splitter".equals(origin)) {
      String hashtag = tuple.getStringByField("hashtag");
      count(hashtags, hashtag, 1);
    } else if("user-hashtag-merger".equals(origin)) {
      String hashtag = tuple.getStringByField("hashtag");
      String user = tuple.getStringByField("user");
      String key = user + ":" + hashtag;
      Integer count = tuple.getIntegerByField("count");
      count(usersHashtags, key, count);
    }
  }


  @Override
  public void finishBatch() {
    String lastCommitedTransaction = jedis.get(LAST_COMMITED_TRANSACTION_FIELD);
    String currentTransaction = ""+id.getTransactionId();

    if(currentTransaction.equals(lastCommitedTransaction))
      return ;

    Transaction multi = jedis.multi();

    multi.set(LAST_COMMITED_TRANSACTION_FIELD, currentTransaction);

    Set<String> keys = hashtags.keySet();
    for (String hashtag : keys) {
      Long count = hashtags.get(hashtag);
      multi.hincrBy("hashtags", hashtag, count);
    }

    keys = users.keySet();
    for (String user : keys) {
      Long count = users.get(user);
      multi.hincrBy("users", user, count);
    }

    keys = usersHashtags.keySet();
    for (String key : keys) {
      Long count = usersHashtags.get(key);
      multi.hincrBy("users_hashtags", key, count);
    }

    multi.exec();
  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
  }
}

This is all very straightforward, but there is a very important detail in the finishBatch method.

...
multi.set(LAST_COMMITED_TRANSACTION_FIELD, currentTransaction);
...

Here you are storing in your database the last transaction ID committed. Why should you do that? Remember that if a transaction fails, Storm will be replaying it as many times as necessary. If you don’t make sure that you already processed the transaction, you could overcount and the whole idea of a transactional topology would be useless. So remember: store the last transaction ID committed and check against it before committing.