The total number of non-repeating words in each tweet

Question

I am new to java and Trident, I imported a project to get tweets, but I want to get something. How does this code get more than one tweet when I received the code that tuple.getValue(0); only the first tweet means?

Problem with me is to get all tweets in a hashset or hashmap to get the total number of distnictive words in each tweet

 public void execute(TridentTuple tuple, TridentCollector collector) { 

this method is used to execute tweet equations

 public Values getValues(Tweet tweet, String[] words){ } 

This code received the first tweet, then received its body, converting it to an array of strings, I know what I need to solve, but could not write it well

My opinion: Do for a loop like

 for (int i=0;i<10;i++) { Tweet tweet = (Tweet) tuple.getValue(i); } 

Answer 1

• For each tweet:
  • For each word on Twitter:
    • Try adding each word to the set.
      If the word already exists in the set, remove it from the set.
  • count the size of the set containing the words for this tweet.

Answer 2

The "problem" is the coincidence of "getting the number of different words across all tweets" and Strom as a stream processor. The query you want to answer can only be computed on a finite set of Tweets. However, when processing a stream, you are processing a potential endless stream of input data.

If you have a finite set of tweets, you can use a batch processing infrastructure such as Flink, Spark, or MapReduce. If you really have an infinite number of tweets, you should rephrase your question ...

As you mentioned, you really want to "iterate over all tweets." Since you are handling the stream this way, there is no such concept. You have an infinite number of input tuples, and Storm applies execute() to each of them (i.e. you can think of it as if Storm "loops over the input" automatically - even in "looping" this is not the right term for This ) Since your calculations are “above all tweets,” you need to maintain a state in your Bolt code so that you can update this state for each tweet. The simple state form in Storm will be a member variable in your Bolt class.

 public class MyBolt implements ??? { // this is your "state" variable private final Set<String> allWords = new HashSet<String>(); public void execute(TridentTuple tuple, TridentCollector collector) { Tweet tweet = (Tweet)tuple.getValue(0); String tweetBody = tweet.getBody(); String words[] = tweetBody.toLowerCase().split(regex); for(String w : words) { // as allWords is a set, you cannot add the same word twice // the second "add" call on the same word will just be ignored // thus, allWords will contain each word exactly once this.allWords.add(w); } } } 

Right now, this code is emitting nothing, because it is not clear what you really want to emit? Like in streaming processing, there is no end, you cannot say "emit the final number of words contained in allWords ". What could you do to emit the current counter after each update ... To do this, add collector.emit(new Values(this.allWords.size())); to the end of execute() .

In addition, I want to add that the presented solution works correctly if parallelism is not applied to MyBolt - otherwise different sets of instances may contain the same word. To resolve this, you would need to tokenize each text in your words in a stateless bolt and move these word streams to the accepted MyBolt , which uses the internal state of Set . The input for MyBolt must also receive data through fieldsGrouping to provide different sets of words for each instance.