Leistungsmerkmale von Pub/Sub-zu-BigQuery-Pipelines

Auf dieser Seite werden die Leistungsmerkmale für Dataflow-Streamingjobs beschrieben, die Daten aus Pub/Sub lesen und in BigQuery schreiben. Außerdem finden Sie Benchmark-Testergebnisse für zwei Arten von Streamingpipelines:

  • Reine Zuordnung (Transformation pro Nachricht): Pipelines, die Transformationen pro Nachricht ausführen, ohne den Status zu verfolgen oder Elemente im Stream zu gruppieren. Beispiele sind ETL, Feldvalidierung und Schemazuordnung.

  • Aggregation mit Zeitfenstern (GroupByKey): Pipelines, die zustandsorientierte Vorgänge ausführen und Daten anhand eines Schlüssels und eines Zeitfensters gruppieren. Beispiele sind das Zählen von Ereignissen, das Berechnen von Summen und das Erfassen von Datensätzen für eine Nutzersitzung.

Die meisten Arbeitslasten für die Streamingdatenintegration fallen in diese beiden Kategorien. Wenn Ihre Pipeline einem ähnlichen Muster folgt, können Sie diese Benchmarks verwenden, um Ihren Dataflow-Job mit einer leistungsstarken Referenzkonfiguration zu vergleichen.

Testmethodik

Die Benchmarks wurden mit den folgenden Ressourcen durchgeführt:

  • Ein vorab bereitgestelltes Pub/Sub-Thema mit einer konstanten Eingabelast. Nachrichten wurden mit der Vorlage „Streaming Data Generator“ generiert.

    • Nachrichtenrate: ca. 1.000.000 Nachrichten pro Sekunde
    • Eingabelast: 1 GiB/s
    • Nachrichtenformat: Zufällig generierter JSON-Text mit einem festen Schema
    • Nachrichtengröße: ca. 1 KiB pro Nachricht

  • Eine Standard BigQuery-Tabelle.

  • Dataflow-Streamingpipelines, die auf der Vorlage „Pub/Sub für BigQuery“ basieren. Diese Pipelines führen die minimal erforderliche Parsing- und Schemazuordnung aus. Es wurde keine benutzerdefinierte Funktion (User-Defined Function, UDF) verwendet.

Nachdem sich die horizontale Skalierung stabilisiert und die Pipeline den stabilen Zustand erreicht hatte, konnten die Pipelines etwa einen Tag lang ausgeführt werden. Anschließend wurden die Ergebnisse erfasst und analysiert.

Dataflow-Pipelines

Es wurden zwei Pipelinevarianten getestet:

Reine Zuordnungspipeline. Diese Pipeline führt eine einfache Zuordnung und Konvertierung von JSON-Nachrichten durch. Für diesen Test wurde die Vorlage „Pub/Sub für BigQuery“ unverändert verwendet.

  • Semantik: Die Pipeline wurde sowohl im „Genau einmal“-Modus als auch im „Mindestens einmal“-Modus getestet. Die „Mindestens einmal“-Verarbeitung bietet einen besseren Durchsatz. Sie sollte jedoch nur verwendet werden, wenn doppelte Datensätze akzeptabel sind oder die nachgelagerte Senke die Deduplizierung übernimmt.

Pipeline mit Aggregation mit Zeitfenstern. Diese Pipeline gruppiert Nachrichten nach einem bestimmten Schlüssel in Fenstern mit fester Größe und schreibt die aggregierten Datensätze in BigQuery. Für diesen Test wurde eine benutzerdefinierte Apache Beam-Pipeline verwendet, die auf der Vorlage „Pub/Sub für BigQuery“ basiert.

  • Aggregationslogik: Für jedes feste, nicht überlappende 1-Minuten-Fenster wurden Nachrichten mit demselben Schlüssel erfasst und als einzelner aggregierter Datensatz in BigQuery geschrieben. Diese Art der Aggregation wird häufig bei der Logverarbeitung verwendet, um zusammengehörige Ereignisse wie die Aktivitäten eines Nutzers zu einem einzelnen Datensatz für die nachgelagerte Analyse zu kombinieren.

  • Schlüsselparallelität: Für den Benchmark wurden 1.000.000 gleichmäßig verteilte Schlüssel verwendet.

  • Semantik: Die Pipeline wurde im „Genau einmal“-Modus getestet. Für Aggregationen ist die „Genau einmal“-Semantik erforderlich, um die Richtigkeit zu gewährleisten und eine Mehrfachzählung innerhalb einer Gruppe und eines Fensters zu verhindern.

Jobkonfiguration

In der folgenden Tabelle sehen Sie, wie die Dataflow-Jobs konfiguriert wurden.

Einstellung Nur Zuordnung, genau einmal Nur Zuordnung, mindestens einmal Aggregation mit Zeitfenstern, genau einmal
Maschinentyp des Workers n1-standard-2 n1-standard-2 n1-standard-2
Worker-vCPUs 2 2 2
Worker-Arbeitsspeicher 7,5 GiB 7,5 GiB 7,5 GiB
Nichtflüchtiger Speicher des Workers Nichtflüchtiger Standardspeicher (HDD), 30 GB Nichtflüchtiger Standardspeicher (HDD), 30 GB Nichtflüchtiger Standardspeicher (HDD), 30 GB
Anfängliche Worker 70 30 180
Höchstanzahl von Workern 100 100 250
Streaming Engine Ja Ja Ja
Horizontales Autoscaling Ja Ja Ja
Abrechnungsmodell Ressourcenbasierte Abrechnung Ressourcenbasierte Abrechnung Ressourcenbasierte Abrechnung
Storage Write API (gRPC) aktiviert? Ja Ja Ja
Storage Write API (gRPC)-Streams 200 Nicht zutreffend 500
Storage Write API (gRPC)-Triggerhäufigkeit 5 Sekunden Nicht zutreffend 5 Sekunden

Die BigQuery Storage Write API (gRPC) wird für Streamingpipelines empfohlen. Wenn Sie den „Genau einmal“-Modus mit der Storage Write API (gRPC) verwenden, können Sie die folgenden Einstellungen anpassen:

  • Anzahl der Schreibstreams. Um eine ausreichende Schlüsselparallelität in der Schreibphase zu gewährleisten, legen Sie die Anzahl der Storage Write API (gRPC)-Streams auf einen Wert fest, der größer als die Anzahl der Worker-CPUs ist. Gleichzeitig sollte ein angemessener Durchsatz für BigQuery-Schreibstreamsbeibehalten werden.

  • Triggerhäufigkeit. Für Pipelines mit hohem Durchsatz ist ein Wert im einstelligen Sekundenbereich geeignet.

Weitere Informationen finden Sie unter Aus Dataflow in BigQuery schreiben.

Benchmark-Ergebnisse

In diesem Abschnitt werden die Ergebnisse der Benchmark-Tests beschrieben.

Durchsatz und Ressourcennutzung

In der folgenden Tabelle sind die Testergebnisse für den Pipeline-Durchsatz und die Ressourcennutzung aufgeführt.

Ergebnis Nur Zuordnung, genau einmal Nur Zuordnung, mindestens einmal Aggregation mit Zeitfenstern, genau einmal
Eingabedurchsatz pro Worker Durchschnitt: 17 MB/s, n=3 Durchschnitt: 21 MB/s, n=3 Durchschnitt: 6 MB/s, n=3
Durchschnittliche CPU-Auslastung aller Worker Durchschnitt: 65%, n=3 Durchschnitt: 69%, n=3 Durchschnitt: 80%, n=3
Anzahl der Worker-Knoten Durchschnitt: 57, n=3 Durchschnitt: 48, n=3 Durchschnitt: 169, n=3
Streaming Engine-Recheneinheiten pro Stunde Durchschnitt: 125, n=3 Durchschnitt: 46, n=3 Durchschnitt: 354, n=3

Der Autoscaling-Algorithmus kann sich auf die Ziel-CPU-Auslastung auswirken. Wenn Sie eine höhere oder niedrigere Ziel-CPU-Auslastung erreichen möchten, können Sie den Autoscaling-Bereich oder den Hinweis zur Worker-Auslastungfestlegen. Höhere Auslastungsziele können zu niedrigeren Kosten, aber auch zu einer schlechteren Tail-Latenz führen, insbesondere bei variablen Lasten.

Bei einer Pipeline mit Aggregation mit Zeitfenstern können sich die Art der Aggregation, die Fenstergröße und die Schlüsselparallelität erheblich auf die Ressourcennutzung auswirken.

Latenz

In der folgenden Tabelle sind die Benchmark-Ergebnisse für die Pipeline-Latenz aufgeführt.

Gesamt-End-to-End-Latenz der Phase Nur Zuordnung, genau einmal Nur Zuordnung, mindestens einmal Aggregation mit Zeitfenstern, genau einmal
P50 Durchschnitt: 800 ms, n=3 Durchschnitt: 160 ms, n=3 Durchschnitt: 3.400 ms, n=3
P95 Durchschnitt: 2.000 ms, n=3 Durchschnitt: 250 ms, n=3 Durchschnitt: 13.000 ms, n=3
P99 Durchschnitt: 2.800 ms, n=3 Durchschnitt: 410 ms, n=3 Durchschnitt: 25.000 ms, n=3

Bei den Tests wurde die End-to-End-Latenz pro Phase (der job/streaming_engine/stage_end_to_end_latencies Messwert) in drei lang andauernden Testausführungen gemessen. Dieser Messwert gibt an, wie viel Zeit die Streaming Engine in jeder Pipelinephase verbringt. Er umfasst alle internen Schritte der Pipeline, z. B.:

  • Shuffling und Warteschlangen für Nachrichten zur Verarbeitung
  • Die tatsächliche Verarbeitungszeit, z. B. das Konvertieren von Nachrichten in Zeilenobjekte
  • Schreiben des nichtflüchtigen Zustands sowie die Zeit, die in der Warteschlange verbracht wird, um den nichtflüchtigen Zustand zu schreiben

Ein weiterer Messwert für die Latenz ist die Datenaktualität. Die Datenaktualität wird jedoch von Faktoren wie der benutzerdefinierten Zeitfensterung und Verzögerungen in der Quelle beeinflusst. Die Systemlatenz bietet eine objektivere Baseline für die interne Verarbeitungseffizienz und den Zustand einer Pipeline unter Last.

Die Daten wurden pro Ausführung etwa einen Tag lang gemessen. Die anfänglichen Startzeiten wurden verworfen, um eine stabile Leistung im stabilen Zustand widerzuspiegeln. Die Ergebnisse zeigen zwei Faktoren, die zusätzliche Latenz verursachen:

  • „Genau einmal“-Modus. Um die „Genau einmal“-Semantik zu erreichen, sind deterministisches Shuffling und Lookups des nichtflüchtigen Zustands für die Deduplizierung erforderlich. Der „Mindestens einmal“-Modus ist deutlich schneller, da diese Schritte übersprungen werden.

  • Aggregation mit Zeitfenstern. Nachrichten müssen vollständig gemischt, gepuffert und in den nichtflüchtigen Zustand geschrieben werden, bevor das Fenster geschlossen wird. Dies erhöht die End-to-End-Latenz.

Die hier gezeigten Benchmarks stellen eine Baseline dar. Die Latenz ist sehr stark von der Komplexität der Pipeline abhängig. Benutzerdefinierte UDFs, zusätzliche Transformationen und eine komplexe Zeitfensterlogik können die Latenz erhöhen. Einfache, stark reduzierende Aggregationen wie Summe und Anzahl führen in der Regel zu einer geringeren Latenz als zustandsintensive Vorgänge wie das Erfassen von Elementen in einer Liste.

Kosten schätzen

Sie können die Baseline-Kosten Ihrer eigenen, vergleichbaren Pipeline mit ressourcenbasierter Abrechnung mit dem Google Cloud Preisrechner schätzen. Gehen Sie dazu so vor:

  1. Öffnen Sie den Preisrechner.
  2. Klicken Sie auf Der Schätzung hinzufügen.
  3. Wählen Sie Dataflow aus.
  4. Wählen Sie für Diensttyp die Option "Dataflow Classic" aus.
  5. Wählen Sie Erweiterte Einstellungen aus, um alle Optionen zu sehen.
  6. Wählen Sie den Standort aus, an dem der Job ausgeführt wird.
  7. Wählen Sie für Jobtyp die Option „Streaming“ aus.
  8. Wählen Sie Streaming Engine aktivieren aus.
  9. Geben Sie Informationen zu den Jobausführungsstunden, Worker-Knoten, Worker-Maschinen und zum nichtflüchtigen Speicherplatz ein.
  10. Geben Sie die geschätzte Anzahl der Streaming Engine-Recheneinheiten ein.

Die Ressourcennutzung und die Kosten skalieren ungefähr linear mit dem Eingabedurchsatz. Bei kleinen Jobs mit nur wenigen Workern werden die Gesamtkosten jedoch von den Fixkosten dominiert. Als Ausgangspunkt können Sie die Anzahl der Worker-Knoten und den Ressourcenverbrauch aus den Benchmark-Ergebnissen ableiten.

Angenommen, Sie führen eine reine Zuordnungspipeline im „Genau einmal“-Modus mit einer Eingabedatenrate von 100 MiB/s aus. Anhand der Benchmark-Ergebnisse für eine 1-GiB/s-Pipeline können Sie die Ressourcenanforderungen so schätzen:

  • Skalierungsfaktor: (100 MiB/s) / (1 GiB/s) = 0,1
  • Voraussichtliche Worker-Knoten: 57 Worker × 0,1 = 5,7 Worker
  • Voraussichtliche Anzahl der Streaming Engine-Recheneinheiten pro Stunde: 125 × 0,1 = 12,5 Einheiten pro Stunde

Dieser Wert sollte nur als erste Schätzung verwendet werden. Der tatsächliche Durchsatz und die Kosten können je nach Faktoren wie Maschinentyp, Verteilung der Nachrichtengröße, Nutzercode, Aggregationstyp, Schlüsselparallelität und Fenstergröße erheblich variieren. Weitere Informationen finden Sie unter Best Practices für die Kostenoptimierung von Dataflow.

Testpipeline ausführen

In diesem Abschnitt werden die gcloud dataflow flex-template run Befehle gezeigt, die zum Ausführen der reinen Zuordnungspipeline verwendet wurden.

„Genau einmal“-Modus 

gcloud dataflow flex-template run JOB_ID \
  --template-file-gcs-location gs://dataflow-templates-us-central1/latest/flex/PubSub_to_BigQuery_Flex \
  --enable-streaming-engine \
  --num-workers 70 \
  --max-workers 100 \
  --parameters \
inputSubscription=projects/PROJECT_IDsubscriptions/SUBSCRIPTION_NAME,\
outputTableSpec=PROJECT_ID:DATASET.TABLE_NAME,\
useStorageWriteApi=true,\
numStorageWriteApiStreams=200 \
storageWriteApiTriggeringFrequencySec=5

„Mindestens einmal“-Modus 

gcloud dataflow flex-template run JOB_ID \
  --template-file-gcs-location gs://dataflow-templates-us-central1/latest/flex/PubSub_to_BigQuery_Flex \
  --enable-streaming-engine \
  --num-workers 30 \
  --max-workers 100 \
  --parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/SUBSCRIPTION_NAME,\
outputTableSpec=PROJECT_ID:DATASET.TABLE_NAME,\
useStorageWriteApi=true \
  --additional-experiments streaming_mode_at_least_once

Ersetzen Sie Folgendes:

  • JOB_ID: die Dataflow-Job-ID
  • PROJECT_ID: die Projekt-ID
  • SUBSCRIPTION_NAME: der Name des Pub/Sub-Abos
  • DATASET: der Name des BigQuery-Datasets
  • TABLE_NAME: der Name der BigQuery-Tabelle

Testdaten erzeugen

Verwenden Sie den folgenden Befehl, um Testdaten zu generieren und die Vorlage „Streaming Data Generator“ auszuführen:

gcloud dataflow flex-template run JOB_ID \
  --template-file-gcs-location gs://dataflow-templates-us-central1/latest/flex/Streaming_Data_Generator \
  --num-workers 70 \
  --max-workers 100 \
  --parameters \
topic=projects/PROJECT_ID/topics/TOPIC_NAME,\
qps=1000000,\
maxNumWorkers=100,\
schemaLocation=SCHEMA_LOCATION

Ersetzen Sie Folgendes:

  • JOB_ID: die Dataflow-Job-ID
  • PROJECT_ID: die Projekt-ID
  • TOPIC_NAME: der Name des Pub/Sub-Themas
  • SCHEMA_LOCATION: der Pfad zu einer Schemadatei in Cloud Storage

Die Vorlage „Streaming Data Generator“ verwendet eine JSON-Datengenerator-Datei, um das Nachrichtenschema zu definieren. Bei den Benchmark-Tests wurde ein Nachrichtenschema ähnlich dem folgenden verwendet:

{
  "logStreamId": "{{integer(1000001,2000000)}}",
  "message": "{{alphaNumeric(962)}}"
}

Nächste Schritte