I tried to enhance the Flink example displaying the usage of streams.
My goal is to use the windowing features (see the window
function call).
I assume that the code below outputs the sum of last 3 numbers of the stream.
(the stream is opened thanks to nc -lk 9999
on ubuntu)
Actually, the output sums up ALL numbers entered. Switching to a time window produces the same result, i.e. no windowing produced.
Is that a bug? (version used: latest master on github )
object SocketTextStreamWordCount {
def main(args: Array[String]) {
val hostName = args(0)
val port = args(1).toInt
val env = StreamExecutionEnvironment.getExecutionEnvironment
// Create streams for names and ages by mapping the inputs to the corresponding objects
val text = env.socketTextStream(hostName, port)
val currentMap = text.flatMap { (x:String) => x.toLowerCase.split("\\W+") }
.filter { (x:String) => x.nonEmpty }
.window(Count.of(3)).every(Time.of(1, TimeUnit.SECONDS))
// .window(Time.of(5, TimeUnit.SECONDS)).every(Time.of(1, TimeUnit.SECONDS))
.map { (x:String) => ("not used; just to have a tuple for the sum", x.toInt) }
val numberOfItems = currentMap.count
numberOfItems print
val counts = currentMap.sum( 1 )
counts print
env.execute("Scala SocketTextStreamWordCount Example")
}
}