Denken Sie daran, ich habe einen Job wie folgt in Spark;
CSV-Datei == & gt; Nach einer Spalte filtern == & gt; Beispiel nehmen == & gt; Als JSON speichern
Nun muss ich wissen, welchen Schritt ( Abrufen der Datei oder Filtern oder Sampling ) des Jobs derzeit programmatisch ausgeführt wird ( Vorzugsweise mit Java API)? Gibt es einen Weg dafür?
Ich kann Job, Bühne und Aufgabe mit der Klasse SparkListener verfolgen. Und es kann wie das Verfolgen einer Bühnen-ID erfolgen. Aber wie kann ich wissen, welche ID für welchen Schritt in der Jobkette gilt?
Ich möchte eine Benachrichtigung an den Nutzer senden, wenn die Option Nach Spalte filtern abgeschlossen ist. Dafür habe ich eine Klasse erstellt, die die SparkListener-Klasse erweitert. Aber ich kann nicht herausfinden, woher ich den Namen des gerade ausgeführten Transformationsnamens bekommen kann. Ist es möglich, überhaupt zu verfolgen?
%Vor%Sie können nicht genau wissen, wann beispielsweise der Filtervorgang beginnt oder endet.
Das liegt daran, dass Sie Transformationen ( filter , map , ...) und Aktionen ( count , foreach , ...) haben. Spark wird so viele Operationen wie möglich in eine Phase einfügen. Dann wird die Bühne parallel auf den verschiedenen Partitionen Ihrer Eingabe ausgeführt. Und hier kommt das Problem.
Angenommen, Sie haben mehrere Worker und das folgende Programm
LOAD == & gt; MAP == & gt; FILTER == & gt; GROUP BY + Aggregation
Dieses Programm wird wahrscheinlich zwei Stufen haben: Die erste Stufe lädt die Datei und wendet map und filter an.
Dann wird die Ausgabe gemischt, um die Gruppen zu erstellen. In der zweiten Stufe wird die Aggregation durchgeführt.
Nun besteht das Problem darin, dass Sie mehrere Arbeiter haben und jeder einen Teil Ihrer Eingabedaten parallel verarbeitet. Das heißt, jeder Executor in Ihrem Cluster erhält eine Kopie Ihres Programms (die aktuelle Phase) und führt dies auf der zugewiesenen Partition aus.
Sie sehen mehrere Instanzen Ihrer Operatoren map und filter , die parallel, aber nicht notwendigerweise gleichzeitig ausgeführt werden. Im Extremfall wird Arbeiter 1 mit Stufe 1 abschließen, bevor Arbeiter 20 überhaupt angefangen hat (und daher mit seiner Operation filter vor Arbeiter 20 enden).
Für RDDs Spark verwendet die Iterator-Modell in einer Phase. Für Datasets in der neuesten Spark-Version erstellen sie jedoch eine einzige Schleife über die Partition und führen die Transformationen aus. Dies bedeutet, dass Spark in diesem Fall nicht wirklich weiß, wann ein Transformationsoperator für eine einzelne Aufgabe fertig ist!
Lange Rede kurzer Sinn:
Also, jetzt hatte ich schon das gleiche Problem:
In unserem Ferkel-Projekt (bitte erlauben Sie etwas Werbung ;-)) generieren wir Spark-Code aus Pig Latin-Skripten und wollten es um die Skripte zu profilieren. Ich habe schließlich mapPartition operator zwischen allen Benutzeroperatoren eingefügt, die die Partitions-ID und die aktuelle Zeit an einen Server senden, der die Nachrichten auswerten wird. Allerdings hat diese Lösung auch ihre Grenzen ... und ich bin noch nicht ganz zufrieden.
Wenn Sie jedoch die Programme nicht ändern können, fürchte ich, dass Sie nicht erreichen können, was Sie wollen.
Haben Sie diese Option in Erwägung gezogen: Ссылка
Es scheint, dass Sie die folgende Ruhe-API verwenden können, um einen bestimmten Job-Status / Anwendungen / [App-ID] / Jobs / [Job-ID]
Sie können JobGroupId und JobGroupDescription so einstellen, dass Sie verfolgen können, welche Jobgruppe bearbeitet wird. dh setJobGroup
Angenommen, Sie rufen den JobGroupId "Test"
auf %Vor%Wenn Sie die Ссылка aufrufen
Sie erhalten einen JSON mit einem aussagekräftigen Namen für diesen Job:
%Vor%Tags und Links scala java apache-spark bigdata