Auf dieser Seite finden Sie Anleitungen zu den Best Practices, die Sie beim Erstellen und Ausführen von hochgradig parallelen Dataflow-HPC-Workflows befolgen sollten. Dazu gehört auch die Verwendung von externem Code in Ihren Pipelines, die Ausführung der Pipeline und die Verwaltung der Fehlerbehandlung.
Externen Code in die Pipeline einbinden
Ein wichtiges Unterscheidungsmerkmal für hochparallele Pipelines besteht darin, dass sie C++-Code in DoFn statt in einer der standardmäßigen Apache Beam SDK-Sprachen verwenden. Für Java-Pipelines wird die Verwendung von externen Prozeduraufrufen empfohlen, um die Verwendung von C++-Bibliotheken in der Pipeline zu vereinfachen. In diesem Abschnitt wird ein allgemeiner Ansatz zum Ausführen von externem Code (C++) in Java-Pipelines beschrieben.
Eine Apache Beam-Pipelinedefinition besteht aus mehreren Schlüsselkomponenten:
PCollectionssind unveränderliche Sammlungen von homogenen Elementen.PTransformswerden verwendet, um die Transformationen in einePCollectionzu definieren, die eine weiterePCollectiongeneriert.- Die Pipeline ist das Konstrukt, mit dem Sie über Code die Interaktionen zwischen
PTransformsundPCollectionsdeklarieren können. Die Pipeline wird als gerichteter azyklischer Graph (DAG) dargestellt.
Wenn Sie Code aus einer Sprache verwenden, die nicht in der Apache Beam SDK-Standardsprache vorliegt, platzieren Sie den Code in PTransform, der sich innerhalb von DoFn befindet, und verwenden Sie eine der Standard-SDK-Sprachen für die Definition der Pipeline selbst.
Wir empfehlen die Verwendung des Apache Beam Python SDK zur Definition der Pipeline, da das Python SDK eine Dienstprogrammklasse hat, die die Verwendung von anderem Code vereinfacht. Sie können jedoch die anderen Apache Beam SDKs verwenden.
Sie können den Code für eigene Experimente verwenden, ohne dass ein vollständiger Build erforderlich ist. Für ein Produktionssystem erstellen Sie in der Regel eigene Binärdateien, um den Ablauf auf Ihre Anforderungen abzustimmen.
Das folgende Diagramm veranschaulicht die zwei Verwendungszwecke von Pipeline-Daten:
- Daten werden verwendet, um den Prozess zu steuern.
- Daten werden während der Verarbeitung erfasst und mit den Treiberdaten verknüpft.
Auf dieser Seite werden primäre Daten (aus der Quelle) als Ausgangsdaten und sekundäre Daten (aus der Verarbeitungsphase) als Zugangsdaten bezeichnet.
In einer Finanzanwendung könnten zum Beispiel ein paar Hunderttausend Handelstransaktionen die Ausgangsdaten sein. Jede Transaktion muss in Verbindung mit Marktdaten verarbeitet werden. Die Marktdaten sind also die Zugangsdaten. In einer Medienanwendung könnten die zu verarbeitenden Bilddateien die Ausgangsdaten sein. Wenn dafür keine zusätzlichen Datenquellen erforderlich sind, gibt es eben keine Zugangsdaten.
Zur Größe der Ausgangsdaten
Wenn das Ausgangsdatenelement im unteren Megabytebereich rangiert, empfiehlt sich die Verwendung des normalen Apache Beam-Paradigmas, das die Erstellung eines PCollection-Objekts aus der Quelle vorsieht, das anschließend zur Verarbeitung an die Apache Beam-Transformationen übergeben wird.
Liegt das Ausgangsdatenelement medientypisch im hohen Megabyte- oder gar Gigabytebereich, können Sie die Ausgangsdaten in Cloud Storage ablegen. Im ersten PCollection-Objekt verwenden Sie dann den Speicher-URI als Referenz, sodass nur der URI auf die Daten verweist.
Zur Größe der Zugangsdaten
Wenn die Zugangsdaten einige Hundert Megabyte groß sind oder darunter, können Sie die Daten über einen Side Input in die Apache Beam-Transformationen laden. Der Side Input sendet das Datenpaket an jeden Worker, der es benötigt.
Liegen die Zugangsdaten im Gigabyte- oder Terabytebereich, verwenden Sie entweder Bigtable oder Cloud Storage, um sie abhängig vom Datentyp mit den Ausgangsdaten zusammenzuführen. Bigtable eignet sich gut für Finanzszenarien, wenn über eine Schlüssel/Wert-Paar-Zuordnung häufig auf Marktdaten zugegriffen wird. Weitere Informationen zum Entwerfen Ihres Bigtable-Schemas, einschließlich Empfehlungen für die Arbeit mit Zeitachsendaten, finden Sie in der folgenden Bigtable-Dokumentation:
Externen Code ausführen
Externer Code kann in Apache Beam auf viele Arten ausgeführt werden.
Erstellen Sie einen Prozess, der von einem
DoFn-Objekt in einer Dataflow-Transformation aufgerufen wird.JNI mit dem Java SDK verwenden
Erstellen Sie einen Unterprozess direkt aus dem
DoFn-Objekt. Dieser Ansatz ist zwar nicht der effizienteste, aber robust und einfach zu implementieren. Aufgrund der potenziellen Probleme bei der Verwendung von JNI wird auf dieser Seite die Verwendung eines Unterprozessaufrufs veranschaulicht.
Berücksichtigen Sie beim Entwerfen Ihres Workflows die vollständige End-to-End-Pipeline. Jegliche Ineffizienzen bei der Ausführung des Prozesses werden durch die Tatsache ausgeglichen, dass die Datenbewegung von der Quelle bis zur Senke mit einer einzigen Pipeline erfolgt. Wenn Sie den Ansatz mit anderen vergleichen, sollten Sie die End-to-End-Zeiten der Pipeline sowie die End-to-End-Kosten in Betracht ziehen.
Die Binärdateien in die Hosts laden
Wenn Sie eine native Apache Beam-Sprache verwenden, verschiebt das Apache Beam SDK den gesamten erforderlichen Code automatisch an die Worker. Wenn Sie jedoch externen Code aufrufen, müssen Sie den Code manuell verschieben.
So verschieben Sie den Code: Das Beispiel zeigt die Schritte für das Apache Beam Java SDK.
- Speichern Sie den kompilierten externen Code zusammen mit den Versionsinformationen in Cloud Storage.
- Erstellen Sie in der Methode
@Setupeinen synchronisierten Block, der überprüft, ob die Codedatei auf der lokalen Ressource verfügbar ist. Anstelle einer physischen Prüfung können Sie die Verfügbarkeit durch eine statische Variable bestätigen lassen, wenn der erste Thread beendet ist. - Ist die Datei nicht verfügbar, laden Sie sie mithilfe der Cloud Storage-Clientbibliothek aus dem Cloud Storage-Bucket in den lokalen Worker. Für diese Aufgabe empfehlen wir die Verwendung der Apache Beam-Klasse
FileSystems. - Prüfen Sie, nachdem die Datei verschoben wurde, ob das Ausführungsbit auf die Codedatei festgelegt wurde.
- Prüfen Sie in einem Produktionssystem den Hash der Binärdateien, um sicherzugehen, dass die Datei korrekt kopiert wurde.
Hinweis: Sie können auch die Apache Beam-Funktion filesToStage verwenden. Dabei müssen Sie jedoch auf verschiedene Vorteile verzichten, die der Runner bezüglich der automatischen Verpackung und dem Verschieben des Java-Codes bietet. Außerdem müssen Sie, da der Aufruf des Unterprozesses einen absoluten Dateispeicherort benötigt, den Klassenpfad und damit den Speicherort der Datei, die von filesToStage verschoben wird, programmatisch festlegen. Wir raten von diesem Ansatz jedoch ab.
Externe Binärdateien ausführen
Bevor Sie externen Code ausführen können, müssen Sie einen Wrapper dafür erstellen. Den Wrapper schreiben Sie in der gleichen Sprache wie den externen Code (z. B. C++) oder als Shell-Skript. Mit dem Wrapper können Sie Dateihandles übergeben und Optimierungen vornehmen, wie im Abschnitt Verarbeitung für kurze CPU-Zyklen optimieren auf dieser Seite beschrieben. Der Wrapper muss kein Kunstwerk sein. Das folgende Code-Snippet zeigt einen einfachen Wrapper in C++.
int main(int argc, char* argv[])
{
if(argc < 3){
std::cerr << "Required return file and data to process" << '\n';
return 1;
}
std::string returnFile = argv[1];
std::string word = argv[2];
std::ofstream myfile;
myfile.open (returnFile);
myfile << word;
myfile.close();
return 0;
}
Der Code liest zwei Parameter aus der Argumentliste. Der erste Parameter ist der Speicherort der Rückgabedatei, an die die Daten gesendet werden. Der zweite Parameter sind die Daten, die der Code dem Nutzer zurückgibt. In einer echten Implementierung würde dieser Code natürlich mehr als „Hallo Welt“ liefern!
Nachdem Sie den Wrapper-Code geschrieben haben, führen Sie den externen Code aus. Gehen Sie dazu so vor:
- Die Daten an die Binärdateien mit dem externen Code senden.
- Die Binärdateien ausführen, Fehler abfangen und Fehler und Ergebnisse protokollieren.
- Die Logging-Informationen verarbeiten.
- Die Daten aus der Verarbeitung erfassen.
Daten an die Binärdateien übertragen
Übergeben Sie Daten an den C++-Code, um die Ausführung der Bibliothek zu starten. In diesem Schritt können Sie die Dataflow-Integration mit anderen Google Cloud-Tools nutzen. Ein Tool wie Bigtable kann mit sehr großen Datasets umgehen und viele gleichzeitige Zugriffe mit niedriger Latenz bewältigen. Dadurch können Tausende von Kernen gleichzeitig auf das Dataset zugreifen. Außerdem kann Bigtable Daten vorverarbeiten und vorbereiten, anreichern und filtern. All diese Schritte können in Beam-Transformationen vor der Ausführung des externen Codes erledigt werden.
In einem Produktionssystem empfiehlt es sich, einen Protokollpuffer zu verwenden, um die Eingabedaten zu kapseln. Die Eingabedaten können Sie in Byte konvertiert und mit base64 codiert werden, bevor sie an die externe Bibliothek übergeben werden. Es gibt zwei Möglichkeiten, Daten an die externe Bibliothek zu übergeben:
- Kleine Eingabedaten. Für kleine Daten, die die maximale Länge des Systems für ein Befehlsargument nicht überschreiten, übergeben Sie das Argument in Position 2 des mit
java.lang.ProcessBuildererstellten Prozesses. - Große Eingabedaten. Für größere Daten erstellen Sie eine Datei, deren Name eine UUID enthält, und speichern darin die für den Prozess erforderlichen Daten.
C++-Code ausführen, Fehler abfangen und Logging
Das Erfassen und Verarbeiten von Fehlerinformationen ist ein wichtiger Teil Ihrer Pipeline. Die vom Dataflow-Runner verwendeten Ressourcen sind sitzungsspezifisch und es ist oft schwierig, Worker-Logdateien zu prüfen. Alle nützlichen Informationen müssen erfasst und an die Logging-Funktion von Dataflow weitergeleitet werden. Die Logging-Daten müssen in einem oder mehreren Cloud Storage-Buckets gespeichert werden.
Der empfohlene Ansatz besteht darin, stdout und stderr auf Dateien umzuleiten, damit Sie sich nicht um einen möglichen Speichermangel kümmern müssen. Im Dataflow-Runner, der den C++-Code aufruft, können Sie beispielsweise folgende Zeilen einfügen:
Java
import java.lang.ProcessBuilder.Redirect;
...
processbuilder.redirectError(Redirect.appendTo(errfile));
processbuilder.redirectOutput(Redirect.appendTo(outFile));
Python
# Requires Apache Beam 2.34 or later.
stopping_times, bad_values = (
integers
| beam.Map(collatz.total_stopping_time).with_exception_handling(
use_subprocess=True))
# Write the bad values to a side channel.
bad_values | 'WriteBadValues' >> beam.io.WriteToText(
os.path.splitext(output_path)[0] + '-bad.txt')
Logging-Informationen verarbeiten
In vielen Anwendungsfällen müssen Millionen von Elementen verarbeitet werden. Bei erfolgreicher Verarbeitung liefern die Logs nur wenig oder keine werthaltigen Informationen. Daher müssen Sie entscheiden, ob Log-Daten beibehalten oder verworfen werden sollen. Sie haben folgende Alternativen, wenn Sie die Log-Daten nicht behalten möchten:
- Falls die Logs keine nützlichen Daten aus der Elementverarbeitung enthalten, sollten Sie sie löschen.
- Schreiben Sie eine Logik, um die Log-Daten stichprobenweise abzufragen, zum Beispiel immer 10.000 Log-Einträge. Wenn sich eine homogene Verarbeitung abzeichnet (das heißt, viele Code-Iterationen erzeugen im Wesentlichen identische Log-Daten), bietet sich dies als ausgewogene Möglichkeit an, Log-Daten zu speichern und gleichzeitig die Verarbeitung zu optimieren.
Bei einem Fehler kann die Datenmenge in den Logs sehr stark anwachsen. Eine effektive Strategie zum Umgang mit großen Mengen von Fehlerlogdaten besteht darin, die ersten Zeilen des Logeintrags zu lesen und nur diese Zeilen an Cloud Logging zu verschieben. Sie können den Rest der Logdatei in Cloud Storage-Buckets laden. So können Sie sich später die ersten Zeilen der Fehlerlogs ansehen und bei Bedarf auf Cloud Storage verweisen.
Die Überprüfung der Größe der Protokolldatei ist ebenfalls hilfreich. Beträgt sie null, können Sie die Datei ignorieren oder eine einfache Lognachricht ausgeben, dass die Datei keine Daten enthält.
Daten aus der Verarbeitung erfassen
Zur Übergabe des Berechnungsergebnisses an die DoFn-Funktion sollten Sie nicht stdout verwenden. Anderer Code, den Ihr C++-Code aufruft, und sogar Ihr eigener Code, könnten ebenfalls Nachrichten an stdout senden und den stdoutput-Stream, der ansonsten Logging-Daten enthält, vollschreiben. Stattdessen empfiehlt es sich, den C++-Wrapper-Code so zu ändern, dass er einen Parameter akzeptiert, der angibt, wo eine Datei erstellt werden soll, die den Wert speichert. Im Idealfall sollte die Datei mithilfe von Protokollpuffern sprachneutral gespeichert werden, damit der C++-Code ein Objekt an den Java- oder Python-Code zurückgeben kann. Das DoFn-Objekt kann das Ergebnis direkt aus der Datei lesen und die Ergebnisse an seinen eigenen output-Aufruf übergeben.
Die Erfahrung hat gezeigt, dass Einheitentests zur Überprüfung eines Vorgangs sehr hilfreich sein können. Implementieren Sie daher einen Einheitentest, der den Prozess unabhängig von der Dataflow-Pipeline ausführt. Fehler in der Bibliothek lassen sich viel effizienter beheben, wenn sie unabhängig ist und nicht die gesamte Pipeline ausgeführt werden muss.
Designverarbeitung für kurze CPU-Zyklen
Der Aufruf eines Unterprozesses erzeugt einen Overhead. Abhängig von der Arbeitslast sind möglicherweise zusätzliche Schritte nötig, um das Verhältnis zwischen der eigentlichen Arbeit und dem Verwaltungsaufwand für das Starten und Herunterfahren des Prozesses auf ein vernünftiges Maß zu reduzieren.
Im Medienanwendungsfall kann die Größe des Ausgangsdatenelements in den hohen Megabyte oder in den Gigabyte liegen. Daher kann die Verarbeitung jedes Datenelements viele Minuten dauern. In diesem Fall sind die Kosten für den Aufruf des Unterprozesses im Vergleich zur Gesamtverarbeitungszeit unbedeutend. Der beste Ansatz in dieser Situation besteht darin, jedes Element seinen eigenen Prozess starten zu lassen.
In anderen Anwendungsfällen, z. B. für die Finanzen, erfordert die Verarbeitung jedoch sehr kleine CPU-Zeiteinheiten (zehntel Millisekunden). In diesem Fall ist der Overhead für den Aufruf des Unterprozesses überproportional groß. Eine Lösung für dieses Problem besteht darin, mit der GroupByKey-Transformation Apache Beam-Batches zu erstellen, die Batches zwischen 50 und 100 Elementen enthalten, die in den Prozess eingespeist werden. Beispielsweise können Sie so vorgehen:
- Sie erstellen ein Schlüssel/Wert-Paar in einer
DoFn-Funktion. Bei der Verarbeitung von Handelstransaktionen könnten Sie die Transaktionsnummer als Schlüssel verwenden. Wenn Sie keine eindeutige Zahl für den Schlüssel haben, können Sie eine Prüfsumme aus den Daten generieren und mit einer Modulo-Funktion Partitionen mit je 50 Elementen erstellen. - Sie senden den Schlüssel an eine
GroupByKey.create-Funktion, die im Gegenzug eineKV<key,Iterable<data>>-Sammlung mit den 50 Elementen zurückgibt, die Sie dann an den Vorgang übergeben.
Worker-Parallelität beschränken
Wenn Sie mit einer Sprache arbeiten, die nativ im Dataflow-Runner unterstützt wird, müssen Sie sich niemals Gedanken darüber machen, was mit dem Worker geschieht. Dataflow hat viele Prozesse zur Überwachung der Flusssteuerung und Threads im Batch- oder Stream-Modus.
Wenn Sie hingegen mit einer externen Sprache wie C++ arbeiten, seien Sie sich darüber im Klaren, dass der Start von Unterprozessen eine eher ungewöhnliche Vorgehensweise ist. Im Batch-Modus verwendet der Dataflow-Runner im Vergleich zum Streaming-Modus ein kleines Verhältnis aus Working-Threads zu CPUs. Insbesondere im Streaming-Modus empfiehlt es sich, ein Semaphor innerhalb der Klasse zu erstellen, um die Parallelität einzelner Worker direkter zu steuern.
Bei der Medienverarbeitung ist es beispielsweise nicht wünschenswert, Hunderte von Transcodierungselementen von einem einzelnen Worker parallel verarbeiten zu lassen. In solchen Fällen können Sie eine Dienstprogrammklasse erstellen, die der DoFn-Funktion Berechtigungen für die auszuführende Arbeit ausstellt. Mit dieser Klasse können Sie die Worker-Threads in der Pipeline direkt steuern.
Datensenken mit hoher Kapazität in Google Cloud verwenden
Nachdem die Daten verarbeitet wurden, werden sie an eine Datensenke gesendet. Die Senke muss in der Lage sein, die Ergebnismenge zu verarbeiten, die von der Grid-Verarbeitungslösung erzeugt wird.
Im folgenden Diagramm sind einige der in Google Cloud verfügbaren Senken zu sehen, wenn Dataflow eine Grid-Arbeitslast ausführt.
Bigtable, BigQuery und Cloud Pub/Sub können alle mit sehr großen Datenstreams umgehen. Beispielsweise kann jeder Bigtable-Knoten 10.000 Inserts pro Sekunde bis zu einer Größe von 1 KB mit einfacher horizontaler Skalierbarkeit verarbeiten. In diesem Fall kann ein Bigtable-Cluster mit 100 Knoten 1.000.000 Nachrichten pro Sekunde aufnehmen, die vom Dataflow-Grid generiert werden.
Segmentierungen verwalten
Wenn Sie C++-Code in einer Pipeline verwenden, müssen Sie entscheiden, wie Segmentierungen verwaltet werden sollen, da sie nicht lokale Auswirkungen haben, wenn sie nicht ordnungsgemäß behandelt werden. Der Dataflow-Runner erstellt Prozesse in Java, Python oder Go nach Bedarf und weist dann den Prozessen in Form von Bundles zu.
Wenn der Aufruf des C++-Codes mit eng gekoppelten Tools wie JNI oder Cython und dem C++-Prozess erfolgt, der Aufrufprozess und die virtuelle Java-Maschine (JVM) stürzt ebenfalls ab. In diesem Szenario sind fehlerhafte Datenpunkte nicht abgefangen. Um fehlerhafte Datenpunkte abfangen zu können, verwenden Sie eine lockere Kopplung, die fehlerhafte Daten verzweigt und die Pipeline ermöglicht, fortzufahren. Mit ausgereiftem C++-Code, der vollständig über alle Datenvarianten hinweg getestet wird, können Sie jedoch Mechanismen wie Cython verwenden.
Nächste Schritte
Folgen Sie der Anleitung, um eine Pipeline zu erstellen, die benutzerdefinierte Container mit C++-Bibliotheken verwendet.
Sehen Sie sich den Beispielcode für diese Seite im GitHub-Repository von Apache Beam an.