Paralleler Iterator in Scala

Question

Ist es irgendwie möglich, mit Scala's parallelen Sammlungen eine Iteratorohne zu parallelisieren, die es vorher vollständig auswertet?

Hier spreche ich über die Parallelisierung der funktionalen Transformationen auf einem Iterator, nämlich map und flatMap. Ich denke, dies erfordert, einige Elemente der Iterator im Voraus zu bewerten und dann mehr zu berechnen, sobald einige über next verbraucht werden.

Alles, was ich finden könnte, würde erfordern, dass der Iterator bestenfalls zu Iterable oder Stream konvertiert wird. Die Stream wird dann komplett ausgewertet, wenn ich .par darauf anrufe.

Ich begrüße auch Vorschläge zur Umsetzung, wenn dies nicht ohne weiteres zur Verfügung steht. Implementierungen sollten parallel map und flatMap unterstützen.

Answer 1

Ihre beste Wette mit der Standardbibliothek verwendet wahrscheinlich nicht parallel Sammlungen aber concurrent.Future.traverse:

import concurrent._ 
import ExecutionContext.Implicits.global 
Future.traverse(Iterator(1,2,3))(i => Future{ i*i }) 

obwohl ich denke, dass das Start das Ganze ausführen wird so bald wie möglich.

Answer 2

Vom ML, Verfahrgeschwindigkeit Iterator Elemente parallel:

https://groups.google.com/d/msg/scala-user/q2NVdE6MAGE/KnutOq3iT3IJ

zog ich Future.traverse aus einem ähnlichen Grund aus. Für meinen Anwendungsfall, bei dem N Jobs funktionieren, habe ich einen Code eingegeben, um den Ausführungskontext aus der Jobwarteschlange zu speisen.

Mein erster Versuch beinhaltete das Blockieren des Feeder-Threads, aber das riskierte auch das Blockieren von Tasks, die Tasks im Ausführungskontext erzeugen wollten. Was weißt du, Blockieren ist böse.

Answer 3

Es ist ein bisschen schwer, genau zu folgen, was Sie nach, aber vielleicht ist es so etwas wie dieses:

val f = (x: Int) => x + 1 
val s = (0 to 9).toStream map f splitAt(6) match { 
    case (left, right) => left.par; right 
} 

Dies f auf den ersten 6 Elemente parallel und dann wieder einen Strom über den Rest eveluate wird .

Answer 4

Ich weiß, dass dies eine alte Frage ist, aber tut die ParIterator Implementierung in der iterata Bibliothek tun, was Sie gesucht haben?

scala> import com.timgroup.iterata.ParIterator.Implicits._ 
scala> val it = (1 to 100000).toIterator.par().map(n => (n + 1, Thread.currentThread.getId)) 
scala> it.map(_._2).toSet.size 
res2: Int = 8 // addition was distributed over 8 threads