Come si consuma flussi secondari raggruppati con mapAsync in Akka flussi

Question

ho bisogno di fare qualcosa di veramente simile a questo https://github.com/typesafehub/activator-akka-stream-scala/blob/master/src/main/scala/sample/stream/GroupLogFile.scala

il mio problema è che ho un numero imprecisato di gruppi e se il numero di parallelismo del mapAsync è meno del numero dei gruppi e ho ottenuto l'errore nell'ultima lavandino

Abbattere SynchronousFileSink (/Users/sam/dev/projects/akka-streams/target/log-ERROR.txt) causa di un errore a monte (akka.stream.impl.StreamSubscriptionTimeoutSupport $$ anon $ 2)

ho cercato di mettere un buffer in mezzo come suggerito nella guida modello di Akka ruscelli http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/stream-cookbook.html

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

ma con lo stesso risultato

Answer 1

Ampliando commento di jrudolph che è del tutto corretta ..

In questo caso non è necessario un mapAsync. Come esempio di base, supponiamo di avere una fonte di tuple

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

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

val originalSource = Source(data) 

È quindi possibile eseguire un groupBy per creare un Source of Sources

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

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

Ognuna delle Fonti raggruppati possono essere trattati in parallelo con un solo map, non c'è bisogno di niente di speciale. Ecco un esempio di ciascun raggruppamento essendo sommate in un flusso indipendente:

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 
    } 

Ora tutti i "foo" numeri vengono sommati in parallelo con tutti i numeri "bar".

mapAsync viene utilizzato quando si dispone di una funzione incapsulata che restituisce un Future[T] e si sta tentando di emettere un T invece; quale non è il caso nella tua domanda. Inoltre, mapAsync coinvolge waiting for results che non è reactive ...