Examples for the sliding operator #25468
This commit is contained in:
parent
044e706eb9
commit
3ca7cb0f94
3 changed files with 136 additions and 5 deletions
|
|
@ -4,13 +4,9 @@ Provide a sliding window over the incoming stream and pass the windows as groups
|
|||
|
||||
@ref[Simple operators](../index.md#simple-operators)
|
||||
|
||||
@@@div { .group-scala }
|
||||
|
||||
## Signature
|
||||
|
||||
@@signature [Flow.scala](/akka-stream/src/main/scala/akka/stream/scaladsl/Flow.scala) { #sliding }
|
||||
|
||||
@@@
|
||||
@apidoc[Flow.sliding](Flow) { scala="#sliding(n:Int,step:Int):FlowOps.this.Repr[scala.collection.immutable.Seq[Out]]" java="#sliding(int,int)" }
|
||||
|
||||
## Description
|
||||
|
||||
|
|
@ -18,6 +14,37 @@ Provide a sliding window over the incoming stream and pass the windows as groups
|
|||
|
||||
Note: the last window might be smaller than the requested size due to end of stream.
|
||||
|
||||
## Examples
|
||||
|
||||
In this first sample we just see the behavior of the operator itself, first with a window of 2 elements and @scala[the default
|
||||
`step` which is 1]@java[a step value of 1].
|
||||
|
||||
Scala
|
||||
: @@snip [Sliding.scala](/akka-docs/src/test/scala/docs/stream/operators/sourceorflow/Sliding.scala) { #sliding-1 }
|
||||
|
||||
Java
|
||||
: @@snip [Sliding.java](/akka-docs/src/test/java/jdocs/stream/operators/sourceorflow/Sliding.java) { #sliding-1 }
|
||||
|
||||
If the stream stops without having seen enough elements to fill a window, the last window will have as many elements
|
||||
was emitted before the stream ended. Here we also provide a step to move two elements forward for each window:
|
||||
|
||||
Scala
|
||||
: @@snip [Sliding.scala](/akka-docs/src/test/scala/docs/stream/operators/sourceorflow/Sliding.scala) { #sliding-2 }
|
||||
|
||||
Java
|
||||
: @@snip [Sliding.java](/akka-docs/src/test/java/jdocs/stream/operators/sourceorflow/Sliding.java) { #sliding-2 }
|
||||
|
||||
One use case for sliding is to implement a moving average, here we do that with a "period" of `5`:
|
||||
|
||||
Scala
|
||||
: @@snip [Sliding.scala](/akka-docs/src/test/scala/docs/stream/operators/sourceorflow/Sliding.scala) { #moving-average }
|
||||
|
||||
Java
|
||||
: @@snip [Sliding.java](/akka-docs/src/test/java/jdocs/stream/operators/sourceorflow/Sliding.java) { #moving-average }
|
||||
|
||||
Sliding can also be used to do simple windowing, see @ref[splitAfter](splitAfter.md).
|
||||
|
||||
|
||||
## Reactive Streams semantics
|
||||
|
||||
@@@div { .callout }
|
||||
|
|
|
|||
|
|
@ -0,0 +1,56 @@
|
|||
/*
|
||||
* Copyright (C) 2009-2020 Lightbend Inc. <https://www.lightbend.com>
|
||||
*/
|
||||
|
||||
package jdocs.stream.operators.sourceorflow;
|
||||
|
||||
import akka.NotUsed;
|
||||
import akka.actor.typed.ActorSystem;
|
||||
import akka.stream.javadsl.Source;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
public class Sliding {
|
||||
|
||||
private final ActorSystem<Void> system = null;
|
||||
|
||||
public void slidingExample1() {
|
||||
// #sliding-1
|
||||
Source<Integer, NotUsed> source = Source.range(1, 4);
|
||||
source.sliding(2, 1).runForeach(n -> System.out.println(n), system);
|
||||
// prints:
|
||||
// [1, 2]
|
||||
// [2, 3]
|
||||
// [3, 4]
|
||||
// #sliding-1
|
||||
}
|
||||
|
||||
public void slidingExample2() {
|
||||
// #sliding-2
|
||||
Source<Integer, NotUsed> source = Source.range(1, 4);
|
||||
source.sliding(3, 2).runForeach(n -> System.out.println(n), system);
|
||||
// prints:
|
||||
// Vector(1, 2, 3)
|
||||
// [1, 2, 3]
|
||||
// [3, 4] - shorter because stream ended before we got 3 elements
|
||||
// #sliding-2
|
||||
}
|
||||
|
||||
public void slidingExample3() {
|
||||
// #moving-average
|
||||
Source<Integer, NotUsed> numbers = Source.from(Arrays.asList(1, 3, 10, 2, 3, 4, 2, 10, 11));
|
||||
Source<Float, NotUsed> movingAverage =
|
||||
numbers
|
||||
.sliding(5, 1)
|
||||
.map(window -> ((float) window.stream().mapToInt(i -> i).sum()) / window.size());
|
||||
movingAverage.runForeach(n -> System.out.println(n), system);
|
||||
// prints
|
||||
// 3.8 = average of 1, 3, 10, 2, 3
|
||||
// 4.4 = average of 3, 10, 2, 3, 4
|
||||
// 4.2 = average of 10, 2, 3, 4, 2
|
||||
// 4.2 = average of 2, 3, 4, 2, 10
|
||||
// 6.0 = average of 3, 4, 2, 10, 11
|
||||
// #moving-average
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,48 @@
|
|||
/*
|
||||
* Copyright (C) 2019-2020 Lightbend Inc. <https://www.lightbend.com>
|
||||
*/
|
||||
|
||||
package docs.stream.operators.sourceorflow
|
||||
|
||||
import akka.stream.scaladsl.Source
|
||||
import akka.actor.ActorSystem
|
||||
|
||||
object Sliding {
|
||||
implicit val system: ActorSystem = ???
|
||||
|
||||
def slidingExample1(): Unit = {
|
||||
//#sliding-1
|
||||
val source = Source(1 to 4)
|
||||
source.sliding(2).runForeach(println)
|
||||
// prints:
|
||||
// Vector(1, 2)
|
||||
// Vector(2, 3)
|
||||
// Vector(3, 4)
|
||||
//#sliding-1
|
||||
}
|
||||
|
||||
def slidingExample2(): Unit = {
|
||||
//#sliding-2
|
||||
val source = Source(1 to 4)
|
||||
source.sliding(n = 3, step = 2).runForeach(println)
|
||||
// prints:
|
||||
// Vector(1, 2, 3)
|
||||
// Vector(3, 4) - shorter because stream ended before we got 3 elements
|
||||
//#sliding-2
|
||||
}
|
||||
|
||||
def slidingExample3(): Unit = {
|
||||
//#moving-average
|
||||
val numbers = Source(1 :: 3 :: 10 :: 2 :: 3 :: 4 :: 2 :: 10 :: 11 :: Nil)
|
||||
val movingAverage = numbers.sliding(5).map(window => window.sum.toFloat / window.size)
|
||||
movingAverage.runForeach(println)
|
||||
// prints
|
||||
// 3.8 = average of 1, 3, 10, 2, 3
|
||||
// 4.4 = average of 3, 10, 2, 3, 4
|
||||
// 4.2 = average of 10, 2, 3, 4, 2
|
||||
// 4.2 = average of 2, 3, 4, 2, 10
|
||||
// 6.0 = average of 3, 4, 2, 10, 11
|
||||
//#moving-average
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue