Benutzerdefinierten Job mit dem Job-Builder erstellen

Mit dem Job-Builder können Sie benutzerdefinierte Batch- und Streamingjobs für Dataflow erstellen. Sie können Job-Builder-Jobs auch als Apache Beam-YAML -Dateien speichern, um sie freizugeben und wiederzuverwenden.

Neue Pipeline erstellen

So erstellen Sie eine neue Pipeline im Job-Builder:

  1. Rufen Sie in der Google Cloud Console die Seite Jobs auf.

    Zu „Jobs“

  2. Klicken Sie auf „Job aus Builder erstellen“.

  3. Geben Sie unter Jobname einen Namen für den Job ein.

  4. Wählen Sie entweder Batch oder Streaming aus.

  5. Wenn Sie Streaming auswählen, wählen Sie einen Fenstermodus aus. Geben Sie dann eine Spezifikation für das Fenster ein:

    • Festes Fenster: Geben Sie eine Fenstergröße in Sekunden ein.
    • Gleitendes Fenster: Geben Sie eine Fenstergröße und einen Fensterzeitraum in Sekunden ein.
    • Sitzungsfenster: Geben Sie eine Sitzungslücke in Sekunden ein.

    Weitere Informationen zur Fensterverarbeitung finden Sie unter Fenster und Fensterfunktionen.

Fügen Sie der Pipeline als Nächstes Quellen, Transformationen und Senken hinzu, wie in den folgenden Abschnitten beschrieben.

Quelle zur Pipeline hinzufügen

Eine Pipeline muss mindestens eine Quelle haben. Anfangs ist der Job-Builder mit einer leeren Quelle gefüllt. So konfigurieren Sie die Quelle:

  1. Geben Sie im Feld Quellname einen Namen für die Quelle ein oder verwenden Sie den Standardnamen. Der Name wird in der Jobgrafik angezeigt, wenn Sie den Job ausführen.

  2. Wählen Sie in der Liste Quelltyp den Typ der Datenquelle aus.

  3. Geben Sie je nach Quelltyp zusätzliche Konfigurationsinformationen an.

    • Wenn Sie beispielsweise BigQuery auswählen, geben Sie die Tabelle an, aus der gelesen werden soll.
    • Wenn Sie Pub/Sub auswählen, geben Sie ein Nachrichtenschema an. Geben Sie den Namen und den Datentyp jedes Felds ein, das Sie aus Pub/Sub-Nachrichten lesen möchten. Die Pipeline verwirft alle Felder, die nicht im Schema angegeben sind.
    • Wenn Sie Apache Iceberg auswählen, geben Sie die Verbindungsdetails für Ihren Iceberg REST-Katalog (IRC) an, z. B. die Iceberg-Tabellen-ID, den Katalognamen, den Katalogtyp, die Katalog-URI und den Warehouse-Namen.
    • Wenn Sie Delta Lake auswählen, geben Sie die Details für Ihre in Cloud Storage gespeicherte Delta Lake-Tabelle an.

  4. Optional: Bei einigen Quelltypen können Sie auf Quellendaten in der Vorschau ansehen klicken, um eine Vorschau der Quellendaten zu sehen.

Wenn Sie der Pipeline eine weitere Quelle hinzufügen möchten, klicken Sie auf Quelle hinzufügen. Wenn Sie Daten aus mehreren Quellen kombinieren möchten, fügen Sie Ihrer Pipeline eine SQL- oder Join-Transformation hinzu.

Transformation zur Pipeline hinzufügen

Optional können Sie der Pipeline eine oder mehrere Transformationen hinzufügen. Mit den folgenden Transformationen können Sie Daten aus Quellen und anderen Transformationen bearbeiten, aggregieren oder zusammenführen:

Transformationstyp Beschreibung Informationen zur Beam-YAML-Transformation
Filtern (Python) Datensätze mit einem Python-Ausdruck filtern.
SQL-Transformation Datensätze bearbeiten oder mehrere Eingaben mit einer SQL-Anweisung zusammenführen.
Felder zuordnen (Python) Neue Felder oder ganze Datensätze mit Python-Ausdrücken und Funktionen hinzufügen bzw. neu zuordnen.
Felder zuordnen (SQL) Datensatzfelder mit SQL-Ausdrücken hinzufügen oder zuordnen.
YAML-Transformationen:
  1. AssertEqual
  2. AssignTimestamps
  3. Kombinieren
  4. Aufteilen
  5. Filter
  6. Vereinfachen
  7. Beitreten
  8. LogForTesting
  9. MLTransform
  10. MapToFields
  11. PyTransform
  12. WindowInfo

Alle Transformationen aus dem Beam YAML SDK verwenden.

YAML-Transformationskonfiguration: Geben Sie die Konfiguration parameter für die YAML-Transformation als YAML-Zuordnung an. Die Schlüssel/Wert-Paare werden verwendet, um den Konfigurationsabschnitt der resultierenden Beam-YAML-Transformation zu füllen. Die unterstützten Konfigurationsparameter für jeden Transformationstyp finden Sie in der Dokumentation zur Beam-YAML-Transformation. Beispielkonfigurationsparameter:

Kombinieren
group_by:
combine:
Beitreten
type:
equalities:
fields:
Log Datensätze in die Worker-Logs des Jobs schreiben.
Gruppieren nach Datensätze mit Funktionen wie count() und sum() kombinieren.
Beitreten Mehrere Eingaben für gleiche Felder verknüpfen.
Aufteilen Datensätze durch Vereinfachen von Arrayfeldern aufteilen.

So fügen Sie eine Transformation hinzu:

  1. Klicken Sie auf Transformation hinzufügen.

  2. Geben Sie im Feld Transformationsname einen Namen für die Transformation ein oder verwenden Sie den Standardnamen. Der Name wird in der Jobgrafik angezeigt, wenn Sie den Job ausführen.

  3. Wählen Sie in der Liste Transformationstyp den Transformationstyp aus.

  4. Geben Sie je nach Transformationstyp zusätzliche Konfigurationsinformationen an. Wenn Sie beispielsweise Filtern (Python) auswählen, geben Sie einen Python Ausdruck ein, der als Filter verwendet werden soll.

  5. Wählen Sie den Eingabeschritt für die Transformation aus. Der Eingabeschritt ist die Quelle oder Transformation, deren Ausgabe die Eingabe für diese Transformation liefert.

Senke zur Pipeline hinzufügen

Eine Pipeline muss mindestens eine Senke haben. Anfangs ist der Job-Builder mit einer leeren Senke gefüllt. So konfigurieren Sie die Senke:

  1. Geben Sie im Feld Senkenname einen Namen für die Senke ein oder verwenden Sie den Standardnamen. Der Name wird in der Jobgrafik angezeigt, wenn Sie den Job ausführen.

  2. Wählen Sie in der Liste Senkentyp den Senkentyp aus.

  3. Geben Sie je nach Senkentyp zusätzliche Konfigurationsinformationen an. Wenn Sie beispielsweise die BigQuery-Senke auswählen, wählen Sie die BigQuery-Tabelle aus, in die Daten geschrieben werden sollen.

  4. Wählen Sie den Eingabeschritt für die Senke aus. Der Eingabeschritt ist die Quelle oder Transformation, deren Ausgabe die Eingabe für diese Transformation liefert.

  5. Wenn Sie der Pipeline eine weitere Senke hinzufügen möchten, klicken Sie auf Senke hinzufügen.

Pipeline ausführen

Führen Sie die folgenden Schritte aus, um eine Pipeline über den Job-Builder auszuführen:

  1. Optional: Legen Sie Dataflow-Joboptionen fest. Klicken Sie zum Erweitern des Abschnitts „Dataflow-Optionen“ auf den Erweiterungspfeil.

  2. Klicken Sie auf Job ausführen. Der Job-Builder leitet Sie zur Jobgrafik für den gesendeten Job weiter. Mit der Jobgrafik können Sie den Status des Jobs überwachen.

Pipeline vor dem Starten validieren

Bei Pipelines mit komplexer Konfiguration, z. B. Python-Filtern und SQL-Ausdrücken, kann es hilfreich sein, die Pipelinekonfiguration vor dem Starten auf Syntaxfehler zu prüfen. So validieren Sie die Pipelinesyntax:

  1. Klicken Sie auf Validieren , um Cloud Shell zu öffnen und den Validierungsdienst zu starten.
  2. Klicken Sie auf Validierung starten.
  3. Wenn bei der Validierung ein Fehler gefunden wird, wird ein rotes Ausrufezeichen angezeigt.
  4. Beheben Sie alle gefundenen Fehler und überprüfen Sie die Korrekturen, indem Sie auf Validieren klicken. Wenn kein Fehler gefunden wird, wird ein grünes Häkchen angezeigt.

Mit der gcloud CLI ausführen

Sie können Beam-YAML-Pipelines auch mit der gcloud CLI ausführen. So führen Sie eine Job-Builder-Pipeline mit der gcloud CLI aus:

  1. Klicken Sie auf YAML speichern , um das Fenster YAML speichern zu öffnen.

  2. Führen Sie eine der folgenden Aktionen aus:

    • Wenn Sie in Cloud Storage speichern möchten, geben Sie einen Cloud Storage-Pfad ein und klicken Sie auf Speichern.
    • Wenn Sie eine lokale Datei herunterladen möchten, klicken Sie auf Herunterladen.

  3. Führen Sie den folgenden Befehl in der Shell oder im Terminal aus:

      gcloud dataflow yaml run my-job-builder-job --yaml-pipeline-file=YAML_FILE_PATH

    Ersetzen Sie YAML_FILE_PATH durch den Pfad Ihrer YAML-Datei, entweder lokal oder in Cloud Storage.

Nächste Schritte