Add Window Bolt support

heron/examples/src/java/com/twitter/heron/examples/PrinterBolt.java

@@ -0,0 +1,34 @@
+// Copyright 2016 Twitter. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.twitter.heron.examples;
+
+import org.apache.storm.topology.BasicOutputCollector;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseBasicBolt;
+import org.apache.storm.tuple.Tuple;
+
+@SuppressWarnings("serial")
+public class PrinterBolt extends BaseBasicBolt {
+
+  @Override
+  public void execute(Tuple tuple, BasicOutputCollector collector) {
+    System.out.println(tuple);
+  }
+
+  @Override
+  public void declareOutputFields(OutputFieldsDeclarer ofd) {
+  }
+
+}

heron/examples/src/java/com/twitter/heron/examples/RandomIntegerSpout.java

@@ -0,0 +1,69 @@
+// Copyright 2016 Twitter. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.twitter.heron.examples;
+
+import java.util.Map;
+import java.util.Random;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+
+/**
+ * Emits a random integer and a timestamp value (offset by one day),
+ * every 100 ms. The ts field can be used in tuple time based windowing.
+ */
+@SuppressWarnings("serial")
+public class RandomIntegerSpout extends BaseRichSpout {
+  private static final Logger LOG = Logger.getLogger(RandomIntegerSpout.class.getName());
+  private SpoutOutputCollector collector;
+  private Random rand;
+  private long msgId = 0;
+
+  @Override
+  public void declareOutputFields(OutputFieldsDeclarer declarer) {
+    declarer.declare(new Fields("value", "ts", "msgid"));
+  }
+
+  @Override
+  public void open(
+      Map<String, Object> conf, TopologyContext context, SpoutOutputCollector newCollector) {
+    collector = newCollector;
+    rand = new Random();
+  }
+
+  @Override
+  public void nextTuple() {
+    Utils.sleep(100);
+    collector.emit(new Values(rand.nextInt(1000),
+        System.currentTimeMillis() - (24 * 60 * 60 * 1000), ++msgId), msgId);
+  }
+
+  @Override
+  public void ack(Object newMsgId) {
+    LOG.log(Level.FINE, "Got ACK for msgId : " + newMsgId);
+  }
+
+  @Override
+  public void fail(Object newMsgId) {
+    LOG.log(Level.FINE, "Got FAIL for msgId : " + newMsgId);
+  }
+}

heron/examples/src/java/com/twitter/heron/examples/SlidingTupleTsTopology.java

@@ -0,0 +1,60 @@
+// Copyright 2016 Twitter. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.twitter.heron.examples;
+
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.storm.Config;
+import org.apache.storm.LocalCluster;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.topology.base.BaseWindowedBolt;
+import org.apache.storm.utils.Utils;
+
+import static org.apache.storm.topology.base.BaseWindowedBolt.Duration;
+
+/**
+ * Windowing based on tuple timestamp (e.g. the time when tuple is generated
+ * rather than when its processed).
+ */
+// SUPPRESS CHECKSTYLE HideUtilityClassConstructor
+public class SlidingTupleTsTopology {
+  public static void main(String[] args) throws Exception {
+    Logger.getGlobal().setLevel(Level.FINE);
+    TopologyBuilder builder = new TopologyBuilder();
+    BaseWindowedBolt bolt = new SlidingWindowSumBolt()
+        .withWindow(new Duration(5, TimeUnit.SECONDS), new Duration(3, TimeUnit.SECONDS))
+        .withTimestampField("ts")
+        .withLag(new Duration(5, TimeUnit.SECONDS));
+    builder.setSpout("integer", new RandomIntegerSpout(), 1);
+    builder.setBolt("slidingsum", bolt, 1).shuffleGrouping("integer");
+    builder.setBolt("printer", new PrinterBolt(), 1).shuffleGrouping("slidingsum");
+    Config conf = new Config();
+    conf.setDebug(true);
+
+    if (args != null && args.length > 0) {
+      conf.setNumWorkers(1);
+      StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
+    } else {
+      LocalCluster cluster = new LocalCluster();
+      cluster.submitTopology("test", conf, builder.createTopology());
+      Utils.sleep(40000);
+      cluster.killTopology("test");
+      cluster.shutdown();
+    }
+  }
+}

heron/examples/src/java/com/twitter/heron/examples/SlidingWindowSumBolt.java

@@ -0,0 +1,79 @@
+// Copyright 2016 Twitter. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.twitter.heron.examples;
+
+import java.util.List;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseWindowedBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.windowing.TupleWindow;
+
+/**
+ * Computes sliding window sum
+ */
+@SuppressWarnings("serial")
+public class SlidingWindowSumBolt extends BaseWindowedBolt {
+  private static final Logger LOG = Logger.getLogger(SlidingWindowSumBolt.class.getName());
+
+  private int sum = 0;
+  private OutputCollector collector;
+
+  @Override
+  @SuppressWarnings("rawtypes")
+  public void prepare(Map stormConf, TopologyContext context, OutputCollector newCollector) {
+    collector = newCollector;
+  }
+
+  @Override
+  public void execute(TupleWindow inputWindow) {
+            /*
+             * The inputWindow gives a view of
+             * (a) all the events in the window
+             * (b) events that expired since last activation of the window
+             * (c) events that newly arrived since last activation of the window
+             */
+    List<Tuple> tuplesInWindow = inputWindow.get();
+    List<Tuple> newTuples = inputWindow.getNew();
+    List<Tuple> expiredTuples = inputWindow.getExpired();
+
+    LOG.log(Level.FINE, "Events in current window: " + tuplesInWindow.size());
+            /*
+             * Instead of iterating over all the tuples in the window to compute
+             * the sum, the values for the new events are added and old events are
+             * subtracted. Similar optimizations might be possible in other
+             * windowing computations.
+             */
+    for (Tuple tuple : newTuples) {
+      sum += (int) tuple.getValue(0);
+    }
+    for (Tuple tuple : expiredTuples) {
+      sum -= (int) tuple.getValue(0);
+    }
+    collector.emit(new Values(sum));
+  }
+
+  @Override
+  public void declareOutputFields(OutputFieldsDeclarer declarer) {
+    declarer.declare(new Fields("sum"));
+  }
+}

heron/examples/src/java/com/twitter/heron/examples/SlidingWindowTopology.java

@@ -0,0 +1,106 @@
+// Copyright 2016 Twitter. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.twitter.heron.examples;
+
+import java.util.List;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.storm.Config;
+import org.apache.storm.LocalCluster;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.topology.base.BaseWindowedBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+import org.apache.storm.windowing.TupleWindow;
+
+import static org.apache.storm.topology.base.BaseWindowedBolt.Count;
+
+/**
+ * A sample topology that demonstrates the usage of {@link org.apache.storm.topology.IWindowedBolt}
+ * to calculate sliding window sum.
+ */
+// SUPPRESS CHECKSTYLE HideUtilityClassConstructor
+public class SlidingWindowTopology {
+
+  private static final Logger LOG = Logger.getLogger(SlidingWindowTopology.class.getName());
+
+  /**
+   * Computes tumbling window average
+   */
+  @SuppressWarnings("serial")
+  private static class TumblingWindowAvgBolt extends BaseWindowedBolt {
+    private OutputCollector collector;
+
+    @Override
+    @SuppressWarnings("rawtypes")
+    public void prepare(Map stormConf, TopologyContext context, OutputCollector newCollector) {
+      collector = newCollector;
+    }
+
+    @Override
+    public void execute(TupleWindow inputWindow) {
+      int sum = 0;
+      List<Tuple> tuplesInWindow = inputWindow.get();
+      LOG.log(Level.FINE, "Events in current window: " + tuplesInWindow.size());
+      if (tuplesInWindow.size() > 0) {
+        /*
+         * Since this is a tumbling window calculation,
+         * we use all the tuples in the window to compute the avg.
+         */
+        for (Tuple tuple : tuplesInWindow) {
+          sum += (int) tuple.getValue(0);
+        }
+        collector.emit(new Values(sum / tuplesInWindow.size()));
+      }
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+      declarer.declare(new Fields("avg"));
+    }
+  }
+
+
+  public static void main(String[] args) throws Exception {
+    TopologyBuilder builder = new TopologyBuilder();
+    builder.setSpout("integer", new RandomIntegerSpout(), 1);
+    builder.setBolt("slidingsum",
+        new SlidingWindowSumBolt().withWindow(new Count(30), new Count(10)), 1)
+        .shuffleGrouping("integer");
+    builder.setBolt("tumblingavg", new TumblingWindowAvgBolt().withTumblingWindow(new Count(3)), 1)
+        .shuffleGrouping("slidingsum");
+    builder.setBolt("printer", new PrinterBolt(), 1).shuffleGrouping("tumblingavg");
+    Config conf = new Config();
+    conf.setDebug(true);
+    if (args != null && args.length > 0) {
+      conf.setNumWorkers(1);
+      StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
+    } else {
+      LocalCluster cluster = new LocalCluster();
+      cluster.submitTopology("test", conf, builder.createTopology());
+      Utils.sleep(40000);
+      cluster.killTopology("test");
+      cluster.shutdown();
+    }
+  }
+}

heron/storm/src/java/BUILD

@@ -2,6 +2,7 @@ package(default_visibility = ["//visibility:public"])
 
 storm_deps_files = [
     "//heron/api/src/java:api-java",
+    "//heron/proto:proto_topology_java",
     "//heron/common/src/java:basics-java",
     "//heron/simulator/src/java:simulator-java",
     "@com_googlecode_json_simple_json_simple//jar",