Как сгруппировать большой поток в подпотоки - PullRequest
0 голосов
/ 23 октября 2018

Я хочу сгруппировать большие Stream[F, A] в Stream[Stream[F, A]] с не более чем n элементом для внутреннего потока.

Это то, что я сделал, в основном конвейерные порции в Queue[F, Queue[F, Chunk[A]], а затем возвращает очередьэлементы как результат потока.

 implicit class StreamSyntax[F[_], A](s: Stream[F, A])(
    implicit F: Concurrent[F]) {

    def groupedPipe(
      lastQRef: Ref[F, Queue[F, Option[Chunk[A]]]],
      n: Int): Pipe[F, A, Stream[F, A]] = { in =>
      val initQs =
        Queue.unbounded[F, Option[Queue[F, Option[Chunk[A]]]]].flatMap { qq =>
          Queue.bounded[F, Option[Chunk[A]]](1).flatMap { q =>
            lastQRef.set(q) *> qq.enqueue1(Some(q)).as(qq -> q)
          }
        }

      Stream.eval(initQs).flatMap {
        case (qq, initQ) =>
          def newQueue = Queue.bounded[F, Option[Chunk[A]]](1).flatMap { q =>
            qq.enqueue1(Some(q)) *> lastQRef.set(q).as(q)
          }

          val evalStream = {
            in.chunks
              .evalMapAccumulate((0, initQ)) {
                case ((i, q), c) if i + c.size >= n =>
                  val (l, r) = c.splitAt(n - i)
                  q.enqueue1(Some(l)) >> q.enqueue1(None) >> q
                    .enqueue1(None) >> newQueue.flatMap { nq =>
                    nq.enqueue1(Some(r)).as(((r.size, nq), c))
                  }
                case ((i, q), c) if (i + c.size) < n =>
                  q.enqueue1(Some(c)).as(((i + c.size, q), c))
              }
              .attempt ++ Stream.eval {
              lastQRef.get.flatMap { last =>
                last.enqueue1(None) *> last.enqueue1(None)
              } *> qq.enqueue1(None)
            }
          }
          qq.dequeue.unNoneTerminate
            .map(
              q =>
                q.dequeue.unNoneTerminate
                  .flatMap(Stream.chunk)
                  .onFinalize(
                    q.dequeueChunk(Int.MaxValue).unNoneTerminate.compile.drain))
            .concurrently(evalStream)
      }
    }

    def grouped(n: Int) = {
      Stream.eval {
        Queue.unbounded[F, Option[Chunk[A]]].flatMap { empty =>
          Ref.of[F, Queue[F, Option[Chunk[A]]]](empty)
        }
      }.flatMap { ref =>
        val p = groupedPipe(ref, n)
        s.through(p)
      }
    }
  }

Но это очень сложно, есть ли более простой способ?

Ответы [ 2 ]

0 голосов
/ 05 февраля 2019

В дополнение к уже упомянутому chunksN также рассмотрите возможность использования groupWithin (fs2 1.0.1):

def groupWithin[F2[x] >: F[x]](n: Int, d: FiniteDuration)(implicit timer: Timer[F2], F: Concurrent[F2]): Stream[F2, Chunk[O]]

Разделите эти потоки на группы элементов, полученных в пределах временного окна или ограниченных количеством элементов, в зависимости от того, что произойдет раньше.Пустые группы, которые могут возникнуть, если никакие элементы не могут быть извлечены из восходящего потока в заданном временном окне, не будут выдаваться.

Примечание: временное окно начинается каждый раз, когда нисходящие вытягивания.

Я не уверен, почему вы хотите, чтобы это были вложенные потоки, поскольку требуется, чтобы в одном пакете было «не более n элементов», что подразумевает, что вы сохраняететрек конечного числа элементов (именно для этого Chunk).В любом случае, Chunk всегда можно представить как Stream с Stream.chunk:

val chunks: Stream[F, Chunk[O]] = ???
val streamOfStreams:  Stream[F, Stream[F, O]] = chunks.map(Stream.chunk)

Вот полный пример использования groupWithin:

import cats.implicits._
import cats.effect.{ExitCode, IO, IOApp}
import fs2._
import scala.concurrent.duration._

object GroupingDemo extends IOApp {
  override def run(args: List[String]): IO[ExitCode] = {
    Stream('a, 'b, 'c).covary[IO]
      .groupWithin(2, 1.second)
      .map(_.toList)
      .showLinesStdOut
      .compile.drain
      .as(ExitCode.Success)
  }
}

Выходы:

Список ('a,' b)

Список ('c)

0 голосов
/ 23 октября 2018

fs2 имеет chunkN chunkLimit методов, которые могут помочь с группировкой

stream.chunkN(n).map(Stream.chunk)

stream.chunkLimit(n).map(Stream.chunk)

chunkN создает куски размера n до концапоток

chunkLimit разбивает существующие фрагменты и может создавать фрагменты с переменным размером.

scala> Stream(1,2,3).repeat.chunkN(2).take(5).toList
res0: List[Chunk[Int]] = List(Chunk(1, 2), Chunk(3, 1), Chunk(2, 3), Chunk(1, 2), Chunk(3, 1))

scala> (Stream(1) ++ Stream(2, 3) ++ Stream(4, 5, 6)).chunkLimit(2).toList
res0: List[Chunk[Int]] = List(Chunk(1), Chunk(2, 3), Chunk(4, 5), Chunk(6))
...