在GraphStage中实现您自己的，您可以精确控制何时以及如何推/拉元素。
这里有一个例子

class LastElementWithin[A](duration: FiniteDuration) extends GraphStage[FlowShape[A, A]] {

    private val in = Inlet[A]("LastElementWithin.in")
    private val out = Outlet[A]("LastElementWithin.out")

    override val shape: FlowShape[A, A] = FlowShape(in, out)

    private sealed trait CallbackEvent

    private case object Pull extends CallbackEvent

    private case object Push extends CallbackEvent

    private case object Flush extends CallbackEvent

    private case object Finish extends CallbackEvent

    override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
      new TimerGraphStageLogic(shape) with StageLogging {

        setHandlers(
          in = in,
          out = out,
          handler = new AbstractInOutHandler {

            override def onPush(): Unit = asyncCallback.invoke(Push)

            override def onPull(): Unit = asyncCallback.invoke(Pull)

            override def onUpstreamFinish(): Unit = asyncCallback.invoke(Finish)
          }
        )

        private val FlushTimerKey = "Flush"

        protected override def onTimer(timerKey: Any): Unit = {
          if (timerKey == FlushTimerKey) {
            asyncCallback.invoke(Flush)
          }
        }

        private val asyncCallback = createAsyncCallback(new Procedure[CallbackEvent] {

          private var last: Option[A] = None

          override def apply(param: CallbackEvent): Unit = {
            param match {
              case Pull   => onPull()
              case Push   => onPush()
              case Finish => onFinish()
              case Flush  => flush()
            }
          }

          private def onPull(): Unit = {
            if (!isTimerActive(FlushTimerKey)) scheduleOnce(FlushTimerKey, duration)
            if (!hasBeenPulled(in)) pull(in)
          }

          private def onPush(): Unit = {
            last = Some(grab(in))
            pull(in)
          }

          private def onFinish(): Unit = {
            cancelTimer(FlushTimerKey)
            last.foreach(emit(out, _))
            completeStage()
          }

          private def flush(): Unit = {
            if (isAvailable(out)) {
              last.foreach(emit(out, _))
              scheduleOnce(FlushTimerKey, duration)
            }
          }
        })
      }
  }

在流动中运行

implicit val as: ActorSystem = ActorSystem("test")
    val done = Source
      .tick(5.nanoseconds, 3.seconds, 1)
      .scan(0)((a, b) => a + b)
      .wireTap(num => println(s"up   ${"%02d".format(num)}"))
      .via(Flow.fromGraph(new LastElementWithin(7.seconds)))
      .wireTap(num => println(s"down ${"%02d".format(num)}"))
      .toMat(Sink.ignore)(Keep.right)
      .run()

产生

up   00
up   01
up   02
up   03
down 03
up   04
up   05
down 05
up   06
up   07
down 07
up   08
up   09
up   10
down 10

要获取时间窗口内n个元素中的最后一个元素，您可以使用groupedWithin运算符，然后使用map，例如：

Source
    .tick(5.seconds, 1.second, 1)
    .scan(0)((a, b) => a + b)
    .wireTap(num => println(s"up ${"%02d".format(num)}"))
    .groupedWithin(Int.MaxValue, 7.seconds)
    .map(_.last)
    .wireTap(num => println(s"down ${"%02d".format(num)}"))
    .runWith(Sink.ignore)