Ich habe einen kleinen Anwendungsfall in Apache flink, das ist ein Batch-Verarbeitungssystem. Ich muss eine Sammlung von Dateien verarbeiten. Die Verarbeitung jeder Datei muss von einem Computer ausgeführt werden. Ich habe das unter dem Code. Die ganze Zeit ist nur ein Task-Slot belegt und die Dateien werden nacheinander abgearbeitet. Ich habe 6 Knoten (also 6 Task-Manager) und konfiguriert 4 Aufgaben-Slots in jedem Knoten. Also, ich erwarte, dass 24 Dateien gleichzeitig verarbeitet werden.
%Vor%Ich habe flink als ./bin/start-cluster.sh Befehl gestartet und die Web-Benutzeroberfläche zeigt, dass es 6 Task-Manager, 24 Task-Slots hat.
Die Ordner enthalten ungefähr 49 Dateien. Wenn ich mapPartition für diese Sammlung erstelle, erwarte ich, dass 49 parallele Prozesse überspannt sind. Aber in meiner Infrastruktur werden sie alle nacheinander verarbeitet. Dies bedeutet, dass nur ein Computer (ein Taskmanager) alle 49 Dateinamen verarbeitet. Was ich will ist, wie konfiguriert 2 Aufgaben pro Slots, erwarte ich 24 Dateien gleichzeitig verarbeitet werden.
Irgendwelche Hinweise helfen hier sicherlich. Ich habe diese Parameter in der Datei flink-conf.yaml
%Vor%Vielen Dank im Voraus. Kann mir jemand Licht geben, wo ich falsch liege?
Wie David beschrieben hat, besteht das Problem darin, dass env.fromCollection(Iterable[T]) eine DataSource mit einer nicht parallelen InputFormat erstellt. Daher wird DataSource mit einer Parallelität von 1 ausgeführt. Die nachfolgenden Operatoren ( mapPartition ) erben diese Parallelität von der Quelle, so dass sie verkettet werden können (dies erspart uns eine Netzwerkshuffle).
Die Lösung dieses Problems besteht darin, die Quelle DataSet über
oder explizit die gewünschte Parallelität beim nachfolgenden Operator ( mapPartition ) einstellen:
Tags und Links scala apache-flink batch-processing