This is an automated email from the ASF dual-hosted git repository.
nicknezis pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-heron.git
The following commit(s) were added to refs/heads/master by this push:
new fdf3430 Add sum/max/min reducers in Scala (#3133)
fdf3430 is described below
commit fdf3430c51eaeb715530ba03fc030e04adb46b23
Author: Ning Wang <[email protected]>
AuthorDate: Wed Jan 20 00:42:03 2021 -0800
Add sum/max/min reducers in Scala (#3133)
---
.../scala/ScalaWindowedWordCountTopology.scala | 4 +-
.../heron/streamlet/scala/StreamletReducers.scala | 48 +++++++++++++++++++
.../streamlet/scala/StreamletReducersTest.scala | 55 ++++++++++++++++++++++
.../streamlet/scala/impl/StreamletImplTest.scala | 8 ++--
4 files changed, 109 insertions(+), 6 deletions(-)
diff --git
a/examples/src/scala/org/apache/heron/examples/streamlet/scala/ScalaWindowedWordCountTopology.scala
b/examples/src/scala/org/apache/heron/examples/streamlet/scala/ScalaWindowedWordCountTopology.scala
index 483c66a..40d6963 100644
---
a/examples/src/scala/org/apache/heron/examples/streamlet/scala/ScalaWindowedWordCountTopology.scala
+++
b/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}"))
diff --git
a/heron/api/src/scala/org/apache/heron/streamlet/scala/StreamletReducers.scala
b/heron/api/src/scala/org/apache/heron/streamlet/scala/StreamletReducers.scala
new file mode 100644
index 0000000..995b38a
--- /dev/null
+++
b/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)
+}
diff --git
a/heron/api/tests/scala/org/apache/heron/streamlet/scala/StreamletReducersTest.scala
b/heron/api/tests/scala/org/apache/heron/streamlet/scala/StreamletReducersTest.scala
new file mode 100644
index 0000000..2bb5cc7
--- /dev/null
+++
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)
+ }
+}
diff --git
a/heron/api/tests/scala/org/apache/heron/streamlet/scala/impl/StreamletImplTest.scala
b/heron/api/tests/scala/org/apache/heron/streamlet/scala/impl/StreamletImplTest.scala
index d4afd0d..2100e73 100644
---
a/heron/api/tests/scala/org/apache/heron/streamlet/scala/impl/StreamletImplTest.scala
+++
b/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)
