// Path: src/main/java/tutorial/storm/trident/operations/ExtractFollowerClassAndContentName.java
// public class ExtractFollowerClassAndContentName extends BaseFunction {
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         Content content = (Content)tuple.get(0);
//         User user = (User)tuple.get(1);
// 
//         String followerClass = classify(user.getFollowersCount());
// 
//         collector.emit(new Values(followerClass, content.getContentName()));
//     }
// 
//     private String classify(int followersCount) {
//         if (followersCount < 100){
//             return "< 100";
//         } else if (followersCount < 10*1000){
//             return "< 10K";
//         } else if (followersCount < 100*1000){
//             return "< 100K";
//         } else {
//             return ">= 100K";
//         }
//     }
// 
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/OnlyEnglish.java
// public class OnlyEnglish extends BaseFilter {
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         User user = (User)tuple.get(0);
// 
//         return "en".equals(user.getLang());
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/OnlyHashtags.java
// public class OnlyHashtags extends BaseFilter {
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         Content content = (Content)tuple.get(0);
//         return "hashtag".equals(content.getContentType());
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/ParseTweet.java
// public class ParseTweet extends BaseFunction {
//     private static final Logger log = LoggerFactory.getLogger(ParseTweet.class);
// 
//     private ContentExtracter extracter;
// 
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         if(extracter == null) extracter = new ContentExtracter();
// 
//         String rawTweetJson = (String)tuple.get(0);
//         Status parsed = parse(rawTweetJson);
//         User user = parsed.getUser();
// 
//         for (Content content : extracter.extract(parsed)) {
//             collector.emit(new Values(parsed, content, user));
//         }
//     }
// 
//     private Status parse(String rawJson){
//         try {
//             Status parsed = DataObjectFactory.createStatus(rawJson);
//             return parsed;
//         } catch (TwitterException e) {
//             log.warn("Invalid tweet json -> " + rawJson, e);
//             return null;
//         }
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/testutil/SampleTweet.java
// public class SampleTweet {
//     private static final Logger log = LoggerFactory.getLogger(SampleTweet.class);
// 
//     private final Iterator<String> sampleTweet;
// 
//     public SampleTweet() throws IOException {
//         ObjectMapper om = new ObjectMapper();
//         JsonFactory factory = new JsonFactory();
//         ImmutableList.Builder<String> b = ImmutableList.builder();
// 
//         InputStreamReader reader = new InputStreamReader(this.getClass().getResourceAsStream("sample_tweet.json"));
//         try {
//             String tweetArray = CharStreams.toString(reader);
//             ArrayNode parsed = (ArrayNode)om.readTree(tweetArray);
//             for (JsonNode tweet : parsed) {
//                 StringWriter sw = new StringWriter();
//                 om.writeTree(factory.createGenerator(sw), tweet);
//                 b.add(sw.toString());
//             }
//             sampleTweet = Iterators.cycle(b.build());
//         } finally {
//             reader.close();
//         }
//     }
// 
//     public String sampleTweet(){
//         return sampleTweet.next();
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/testutil/TestUtils.java
// public class TestUtils {
//     public static TransactionalTridentKafkaSpout testTweetSpout(BrokerHosts hosts) {
//         TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(hosts, "test", "storm");
//         kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
//         return new TransactionalTridentKafkaSpout(kafkaConfig);
//     }
// 
// }


import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import storm.kafka.trident.TransactionalTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.FirstN;
import storm.trident.operation.builtin.MapGet;
import storm.trident.operation.builtin.TupleCollectionGet;
import storm.trident.testing.FeederBatchSpout;
import storm.trident.testing.MemoryMapState;
import tutorial.storm.trident.operations.ExtractFollowerClassAndContentName;
import tutorial.storm.trident.operations.OnlyEnglish;
import tutorial.storm.trident.operations.OnlyHashtags;
import tutorial.storm.trident.operations.ParseTweet;
import tutorial.storm.trident.testutil.SampleTweet;
import tutorial.storm.trident.testutil.TestUtils;
import java.io.IOException;
package tutorial.storm.trident.example;



/**
* @author Enno Shioji (enno.shioji@peerindex.com)
*/
public class TopHashtagByFollowerClass {

    public static StormTopology buildTopology(TransactionalTridentKafkaSpout spout) throws IOException {

        TridentTopology topology = new TridentTopology();
        TridentState count =
        topology
                .newStream("tweets", spout)
                .each(new Fields("str"), new ParseTweet(), new Fields("text", "content", "user"))
                .project(new Fields("content", "user"))
                .each(new Fields("content"), new OnlyHashtags())
// Path: src/main/java/tutorial/storm/trident/operations/ExtractFollowerClassAndContentName.java
// public class ExtractFollowerClassAndContentName extends BaseFunction {
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         Content content = (Content)tuple.get(0);
//         User user = (User)tuple.get(1);
// 
//         String followerClass = classify(user.getFollowersCount());
// 
//         collector.emit(new Values(followerClass, content.getContentName()));
//     }
// 
//     private String classify(int followersCount) {
//         if (followersCount < 100){
//             return "< 100";
//         } else if (followersCount < 10*1000){
//             return "< 10K";
//         } else if (followersCount < 100*1000){
//             return "< 100K";
//         } else {
//             return ">= 100K";
//         }
//     }
// 
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/OnlyEnglish.java
// public class OnlyEnglish extends BaseFilter {
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         User user = (User)tuple.get(0);
// 
//         return "en".equals(user.getLang());
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/OnlyHashtags.java
// public class OnlyHashtags extends BaseFilter {
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         Content content = (Content)tuple.get(0);
//         return "hashtag".equals(content.getContentType());
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/ParseTweet.java
// public class ParseTweet extends BaseFunction {
//     private static final Logger log = LoggerFactory.getLogger(ParseTweet.class);
// 
//     private ContentExtracter extracter;
// 
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         if(extracter == null) extracter = new ContentExtracter();
// 
//         String rawTweetJson = (String)tuple.get(0);
//         Status parsed = parse(rawTweetJson);
//         User user = parsed.getUser();
// 
//         for (Content content : extracter.extract(parsed)) {
//             collector.emit(new Values(parsed, content, user));
//         }
//     }
// 
//     private Status parse(String rawJson){
//         try {
//             Status parsed = DataObjectFactory.createStatus(rawJson);
//             return parsed;
//         } catch (TwitterException e) {
//             log.warn("Invalid tweet json -> " + rawJson, e);
//             return null;
//         }
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/testutil/SampleTweet.java
// public class SampleTweet {
//     private static final Logger log = LoggerFactory.getLogger(SampleTweet.class);
// 
//     private final Iterator<String> sampleTweet;
// 
//     public SampleTweet() throws IOException {
//         ObjectMapper om = new ObjectMapper();
//         JsonFactory factory = new JsonFactory();
//         ImmutableList.Builder<String> b = ImmutableList.builder();
// 
//         InputStreamReader reader = new InputStreamReader(this.getClass().getResourceAsStream("sample_tweet.json"));
//         try {
//             String tweetArray = CharStreams.toString(reader);
//             ArrayNode parsed = (ArrayNode)om.readTree(tweetArray);
//             for (JsonNode tweet : parsed) {
//                 StringWriter sw = new StringWriter();
//                 om.writeTree(factory.createGenerator(sw), tweet);
//                 b.add(sw.toString());
//             }
//             sampleTweet = Iterators.cycle(b.build());
//         } finally {
//             reader.close();
//         }
//     }
// 
//     public String sampleTweet(){
//         return sampleTweet.next();
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/testutil/TestUtils.java
// public class TestUtils {
//     public static TransactionalTridentKafkaSpout testTweetSpout(BrokerHosts hosts) {
//         TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(hosts, "test", "storm");
//         kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
//         return new TransactionalTridentKafkaSpout(kafkaConfig);
//     }
// 
// }


// Path: src/main/java/tutorial/storm/trident/example/TopHashtagByFollowerClass.java
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import storm.kafka.trident.TransactionalTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.FirstN;
import storm.trident.operation.builtin.MapGet;
import storm.trident.operation.builtin.TupleCollectionGet;
import storm.trident.testing.FeederBatchSpout;
import storm.trident.testing.MemoryMapState;
import tutorial.storm.trident.operations.ExtractFollowerClassAndContentName;
import tutorial.storm.trident.operations.OnlyEnglish;
import tutorial.storm.trident.operations.OnlyHashtags;
import tutorial.storm.trident.operations.ParseTweet;
import tutorial.storm.trident.testutil.SampleTweet;
import tutorial.storm.trident.testutil.TestUtils;
import java.io.IOException;

package tutorial.storm.trident.example;



/**
* @author Enno Shioji (enno.shioji@peerindex.com)
*/
public class TopHashtagByFollowerClass {

    public static StormTopology buildTopology(TransactionalTridentKafkaSpout spout) throws IOException {

        TridentTopology topology = new TridentTopology();
        TridentState count =
        topology
                .newStream("tweets", spout)
                .each(new Fields("str"), new ParseTweet(), new Fields("text", "content", "user"))
                .project(new Fields("content", "user"))
                .each(new Fields("content"), new OnlyHashtags())
                .each(new Fields("user"), new OnlyEnglish())