Skip to content
Permalink
Browse files
Add sum/max/min reducers in Scala (#3133)
  • Loading branch information
nwangtw committed Jan 20, 2021
1 parent 1a0d198 commit fdf3430c51eaeb715530ba03fc030e04adb46b23
Show file tree
Hide file tree
Showing 4 changed files with 109 additions and 6 deletions.
@@ -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}"))
@@ -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)
}
@@ -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)
}
}
@@ -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)

0 comments on commit fdf3430

Please sign in to comment.