Elaborazione parallela di raccolta di dati di dimensioni superiori alla memoria

Question

Esiste un modo semplice per utilizzare le collezioni parallele scala senza caricare una raccolta completa in memoria?

Ad esempio, ho una grande collezione e mi piacerebbe eseguire una particolare operazione (fold) in parallelo solo su un piccolo blocco, che si adatta alla memoria, che su un altro blocco e così via, e infine ricombinare i risultati di tutti i pezzi

So che gli attori potrebbero essere utilizzati, ma sarebbe davvero bello usare le raccolte par.

ho scritto una soluzione, ma non è bello:

def split[A](list: Iterable[A], chunkSize: Int): Iterable[Iterable[A]] = { 
    new Iterator[Iterable[A]] { 
     var rest = list 
     def hasNext = !rest.isEmpty 
     def next = { 
     val chunk = rest.take(chunkSize) 
     rest = rest.drop(chunkSize) 
     chunk 
     } 
    }.toIterable 
    }            

    def foldPar[A](acc: A)(list: Iterable[A], chunkSize: Int, combine: ((A, A) => A)): A = { 
    val chunks: Iterable[Iterable[A]] = split(list, chunkSize) 
    def combineChunk: ((A,Iterable[A]) => A) = { case (res, entries) => entries.par.fold(res)(combine) } 
    chunks.foldLeft(acc)(combineChunk) 
    }            

    val chunkSize = 10000000       
    val x = 1 to chunkSize*10     

    def sum: ((Int,Int) => Int) = {case (acc,n) => acc + n } 

    foldPar(0)(x,chunkSize,sum) 

Answer 1

La tua idea è molto pulito ed è un peccato non esiste una funzione disponibile già (per quanto ne so).

Ho appena riformulato la tua idea in un codice un po 'più breve. Innanzitutto, ritengo che per il folding parallelo sia utile utilizzare il concetto di monoid: è una struttura con un'operazione associativa e un elemento zero. L'associatività è importante perché non conosciamo l'ordine in cui combiniamo i risultati calcolati in parallelo. E l'elemento zero è importante in modo che possiamo dividere i calcoli in blocchi e iniziare a piegarli ognuno dallo zero. Non c'è nulla di nuovo, tuttavia, è proprio quello che si aspetta fold per le collezioni di Scala.

// The function defined by Monoid's apply must be associative 
// and zero its identity element. 
trait Monoid[A] 
    extends Function2[A,A,A] 
{ 
    val zero: A 
} 

successivo, di Scala Iterator s hanno già un metodo utile grouped(Int): GroupedIterator[Seq[A]] che taglia all'iteratore in sequenze di dimensione fissa. È abbastanza simile al tuo split. Ciò ci permette di tagliare l'ingresso in blocchi di dimensione fissa e quindi applicare metodi di raccolta parallele di Scala sopra:

def parFold[A](c: Iterator[A], blockSize: Int)(implicit monoid: Monoid[A]): A = 
    c.grouped(blockSize).map(_.par.fold(monoid.zero)(monoid)) 
         .fold(monoid.zero)(monoid); 

Pieghiamo ciascun blocco utilizzando il framework collezioni parallele e poi (senza parallelizzazione) combinare i risultati intermedi.

Un esempio:

// Example: 
object SumMonoid extends Monoid[Long] { 
    override val zero: Long = 0; 
    override def apply(x: Long, y: Long) = x + y; 
} 
val it = Iterator.range(1, 10000001).map(_.toLong) 
println(parFold(it, 100000)(SumMonoid));