Wie gruppiert Teilströme mit mapAsync in akka verbrauchen Ströme

Question

ich etwas wirklich ähnlich wie diese https://github.com/typesafehub/activator-akka-stream-scala/blob/master/src/main/scala/sample/stream/GroupLogFile.scala

mein Problem tun müssen, ist, dass ich eine unbekannte Anzahl von Gruppen haben und wenn die Anzahl der Parallelität der mapAsync weniger der Anzahl der Gruppen I got und Fehler in der letzten Senke

Reissen unten SynchronousFileSink (/Users/sam/dev/projects/akka-streams/target/log-ERROR.txt) wegen vorgeschalteten Fehler (akka.stream.impl.StreamSubscriptionTimeoutSupport $$ anon $ 2)

Ich habe versucht, einen Puffer in der Mitte zu setzen, wie in der Musterführung von akka

http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/stream-cookbook.html Strömen vorgeschlagen

groupBy { 
    case LoglevelPattern(level) => level 
    case other     => "OTHER" 
}.buffer(1000, OverflowStrategy.backpressure). 
    // write lines of each group to a separate file 
    mapAsync(parallelism = 2) {.... 

aber mit dem gleichen Ergebnis

Answer 1

auf jrudolph Kommentar Erweiterung ist absolut korrekt ist .. .

Sie benötigen in diesem Fall keine mapAsync. Als einfaches Beispiel: Angenommen, Sie eine Quelle von Tupeln haben

import akka.stream.scaladsl.{Source, Sink} 

def data() = List(("foo", 1), 
        ("foo", 2), 
        ("bar", 1), 
        ("foo", 3), 
        ("bar", 2)) 

val originalSource = Source(data) 

Sie können dann eine groupBy durchführen, um eine

Jeder der gruppierten Quellen Source of Sources

def getID(tuple : (String, Int)) = tuple._1 

//a Source of (String, Source[(String, Int),_]) 
val groupedSource = originalSource groupBy getID 

zu erstellen mit nur parallel verarbeitet werden map, keine Notwendigkeit für etwas Phantasie. Hier ist ein Beispiel für jede Gruppierung in einem unabhängigen Strom summiert werden:

import akka.actor.ActorSystem 
import akka.stream.ACtorMaterializer 

implicit val actorSystem = ActorSystem() 
implicit val mat = ActorMaterializer() 
import actorSystem.dispatcher 

def getValues(tuple : (String, Int)) = tuple._2 

//does not have to be a def, we can re-use the same sink over-and-over 
val sumSink = Sink.fold[Int,Int](0)(_ + _) 

//a Source of (String, Future[Int]) 
val sumSource = 
    groupedSource map { case (id, src) => 
    id -> {src map getValues runWith sumSink} //calculate sum in independent stream 
    } 

nun alle der "foo" Zahlen werden parallel zusammengefaßt, mit all den "bar" Zahlen.

mapAsync wird verwendet, wenn Sie eine gekapselte Funktion haben, die Future[T] zurückgibt und Sie versuchen, stattdessen eine T zu emittieren; was bei dir nicht der Fall ist. Außerdem beinhaltet mapAsync waiting for results, was nicht reactive ist ...