Add sum/max/min reducers in Scala (#3133)

Author: nwangtw

examples/src/scala/org/apache/heron/examples/streamlet/scala/ScalaWindowedWordCountTopology.scala

@@ -24,7 +24,7 @@ import scala.util.Random
 
 import org.apache.heron.examples.streamlet.scala.common.ScalaTopologyExampleUtils
 import org.apache.heron.streamlet.{Config, KeyValue, KeyedWindow, WindowConfig}
-import org.apache.heron.streamlet.scala.{Builder, Runner}
+import org.apache.heron.streamlet.scala.{Builder, Runner, StreamletReducers}
 
 /**
   * This topology is an implementation of the classic word count example
@@ -62,7 +62,7 @@ object ScalaWindowedWordCountTopology {
       .reduceByKeyAndWindow[String, Int]((word: String) => word,
                                          (x: String) => 1,
                                          WindowConfig.TumblingCountWindow(50),
-                                         (x: Int, y: Int) => x + y)
+                                         StreamletReducers.sum(_: Int, _: Int))
       .setName("reduce-operation")
       .consume((kv: KeyValue[KeyedWindow[String], Int]) =>
         log.info(s"word: ${kv.getKey.getKey} - count: ${kv.getValue}"))

heron/api/src/scala/org/apache/heron/streamlet/scala/StreamletReducers.scala

@@ -0,0 +1,48 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.heron.streamlet.scala
+
+/**
+ * This class contains a few standard reduces that can be used with
+ * Streamlet reduce functions such as reduceByKeyAndWindow.
+ * Example, assuming s is a Stringlet<T> object and each tuple has these functions:
+ *   - Integer getKey() and
+ *   - Double getValue()
+ * To get streams of sum, min and max of all values upto the current one:
+ *   s.reduceByKey(T::getKey, T::getValue, StreamletReducers::sum);
+ *   s.reduceByKey(T::getKey, T::getValue, StreamletReducers::min);
+ *   s.reduceByKey(T::getKey, T::getValue, StreamletReducers::max);
+ */
+object StreamletReducers {
+
+  def sum(a: Int, b: Int): Int = a + b
+  def sum(a: Long, b: Long): Long = a + b
+  def sum(a: Float, b: Float): Float = a + b
+  def sum(a: Double, b: Double): Double = a + b
+
+  def max(a: Int, b: Int): Int = math.max(a, b)
+  def max(a: Long, b: Long): Long = math.max(a, b)
+  def max(a: Float, b: Float): Float = math.max(a, b)
+  def max(a: Double, b: Double): Double = math.max(a, b)
+
+  def min(a: Int, b: Int): Int = math.min(a, b)
+  def min(a: Long, b: Long): Long = math.min(a, b)
+  def min(a: Float, b: Float): Float = math.min(a, b)
+  def min(a: Double, b: Double): Double = math.min(a, b)
+}

heron/api/tests/scala/org/apache/heron/streamlet/scala/StreamletReducersTest.scala

@@ -0,0 +1,55 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.heron.streamlet.scala
+
+import org.junit.Assert.assertEquals
+
+import org.apache.heron.streamlet.scala.common.BaseFunSuite
+
+class StreamletReducersTest extends BaseFunSuite {
+
+  test("Sum should work correctly") {
+    assertEquals(StreamletReducers.sum(1, 2), 3)
+    assertEquals(StreamletReducers.sum(1L, 2L), 3L)
+    assertEquals(StreamletReducers.sum(1.0f, 2.0f), 3.0f, 0.01f)
+    assertEquals(StreamletReducers.sum(1.0, 2.0), 3.0, 0.01)
+  }
+
+  test("Max should work correctly") {
+    assertEquals(StreamletReducers.max(1, 2), 2)
+    assertEquals(StreamletReducers.max(2, 1), 2)
+    assertEquals(StreamletReducers.max(1L, 2L), 2L)
+    assertEquals(StreamletReducers.max(2L, 1L), 2L)
+    assertEquals(StreamletReducers.max(1.0f, 2.0f), 2.0f, 0.01f)
+    assertEquals(StreamletReducers.max(2.0f, 1.0f), 2.0f, 0.01f)
+    assertEquals(StreamletReducers.max(1.0, 2.0), 2.0, 0.01)
+    assertEquals(StreamletReducers.max(2.0, 1.0), 2.0, 0.01)
+  }
+
+  test("Min should work correctly") {
+    assertEquals(StreamletReducers.min(1, 2), 1)
+    assertEquals(StreamletReducers.min(2, 1), 1)
+    assertEquals(StreamletReducers.min(1L, 2L), 1L)
+    assertEquals(StreamletReducers.min(2L, 1L), 1L)
+    assertEquals(StreamletReducers.min(1.0f, 2.0f), 1.0f, 0.01f)
+    assertEquals(StreamletReducers.min(2.0f, 1.0f), 1.0f, 0.01f)
+    assertEquals(StreamletReducers.min(1.0, 2.0), 1.0, 0.01)
+    assertEquals(StreamletReducers.min(2.0, 1.0), 1.0, 0.01)
+  }
+}

heron/api/tests/scala/org/apache/heron/streamlet/scala/impl/StreamletImplTest.scala

@@ -54,7 +54,7 @@ import org.apache.heron.streamlet.impl.streamlets.{
   UnionStreamlet
 }
 
-import org.apache.heron.streamlet.scala.{Builder, Streamlet}
+import org.apache.heron.streamlet.scala.{Builder, Streamlet, StreamletReducers}
 import org.apache.heron.streamlet.scala.common.{
   BaseFunSuite,
   TestIncrementSerializableTransformer,
@@ -623,7 +623,7 @@ class StreamletImplTest extends BaseFunSuite {
     supplierStreamlet
       .reduceByKey[Int, Int]((x: Int) => x * 100,
                              (x: Int) => x,
-                             (x: Int, y: Int) => x + y)  // sum operation
+                             StreamletReducers.sum(_: Int, _: Int))
       .setName("Reduce_Streamlet_1")
       .setNumPartitions(5)
 
@@ -651,7 +651,7 @@ class StreamletImplTest extends BaseFunSuite {
     supplierStreamlet
       .reduceByKey[Int, Int]((key: Int) => key * 100,
                              0,
-                             (x: Int, y: Int) => x + y)  // sum operation
+                             StreamletReducers.sum(_: Int, _: Int))
       .setName("Reduce_Streamlet_1")
       .setNumPartitions(5)
 
@@ -680,7 +680,7 @@ class StreamletImplTest extends BaseFunSuite {
       .reduceByKeyAndWindow[Int, Int]((key: Int) => key * 100,
                                       (value: Int) => 1,
                                       WindowConfig.TumblingCountWindow(10),
-                                      (x: Int, y: Int) => x + y)
+                                      StreamletReducers.sum(_: Int, _: Int))
       .setName("Reduce_Streamlet_1")
       .setNumPartitions(5)