Leistungsmerkmale von Kafka-zu-BigQuery-Pipelines

Auf dieser Seite werden die Leistungsmerkmale für Dataflow-Streamingjobs beschrieben, die Daten aus Apache Kafka lesen und in BigQuery schreiben. Sie enthält Benchmark-Testergebnisse für reine Zuordnungspipelines, die Transformationen pro Nachricht ausführen, ohne den Status zu verfolgen oder Elemente im Stream zu gruppieren.

Viele Arbeitslasten für die Datenintegration, einschließlich ETL, Feldvalidierung und Schemazuordnung, fallen in die Kategorie „Nur Zuordnung“. Wenn Ihre Pipeline diesem 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 Managed Service for Apache Kafka-Cluster. 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
    • Kafka-Partitionen: 1.000

  • Eine Standard BigQuery-Tabelle.

  • Eine Dataflow-Streamingpipeline, die die Vorlage „Apache Kafka für BigQuery“ verwendet. Diese Pipeline führt die minimal erforderliche Parsing- und Schemazuordnung aus. Es wurde keine benutzerdefinierte Funktion (User-Defined Function, UDF) verwendet.

Nachdem sich das horizontale Autoscaling stabilisiert und die Pipeline einen stabilen Zustand erreicht hatte, konnten die Pipelines etwa einen Tag lang ausgeführt werden. Danach wurden die Ergebnisse erfasst und analysiert.

Dataflow-Pipeline

Dieser Benchmark verwendet eine reine Zuordnungspipeline, die eine einfache Zuordnung und Konvertierung von JSON-Nachrichten ausführt. Die Pipeline wurde sowohl im Modus „Genau einmal“ als auch im Modus „Mindestens einmal“ getestet. Die Verarbeitung „Mindestens einmal“ bietet einen besseren Durchsatz. Sie sollte jedoch nur verwendet werden, wenn doppelte Datensätze akzeptabel sind oder die nachgelagerte Senke die Deduplizierung übernimmt.

Jobkonfiguration

In der folgenden Tabelle ist die Konfiguration der Dataflow-Jobs dargestellt.

Einstellung Wert
Maschinentyp des Workers e2-standard-2
Worker-Maschinen-vCPUs 2
Worker-Maschinen-RAM 8 GB
Nichtflüchtiger Speicher der Worker-Maschine Nichtflüchtiger Standardspeicher (HDD), 30 GB
Höchstanzahl von Workern 120
Streaming Engine Ja
Horizontales Autoscaling Ja
Abrechnungsmodell Ressourcenbasierte Abrechnung
Storage Write API (gRPC) aktiviert? Ja
Storage Write API (gRPC)-Streams 400
Storage Write API (gRPC)-Triggerhäufigkeit 5 Sekunden
Nachrichtenformat JSON
Kafka-Authentifizierungsmodus

Standardanmeldedaten für Anwendungen (Application Default Credentials, ADC).

Weitere Informationen finden Sie unter Authentifizierungstypen für Kafka-Broker.

Die BigQuery Storage Write API (gRPC) wird für Streamingpipelines empfohlen. Wenn Sie den Modus „Genau einmal“ 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. Beachten Sie dabei die Empfehlungen zum Durchsatz pro Stream.

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

Weitere Informationen finden Sie unter Aus Dataflow in BigQuery schreiben.

Auch die Anzahl der Apache Kafka-Partitionen sollte berücksichtigt werden. Um eine ausreichende Schlüsselparallelität in der Lesestufe zu gewährleisten, sollte die Anzahl der Partitionen mindestens der Gesamtzahl der Worker-vCPUs entsprechen. Weitere Informationen finden Sie unter Daten aus Apache Kafka in Dataflow lesen.

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 Genau einmal Mindestens einmal
Eingabedurchsatz pro Worker Durchschnitt: 15 MB/s, n=3 Durchschnitt: 18 MB/s, n=3
Durchschnittliche CPU-Auslastung aller Worker Durchschnitt: 70%, n=3 Durchschnitt: 75%, n=3
Anzahl der Worker-Knoten Durchschnitt: 63, n=3 Durchschnitt: 53, n=3
Streaming Engine-Recheneinheiten pro Stunde Durchschnitt: 58, n=3 Durchschnitt: 0, 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.

Latenz

In der folgenden Tabelle sind die Benchmark-Ergebnisse für die Pipeline-Latenz im Modus „Genau einmal“ aufgeführt, wobei die Eingabestufe ausgeschlossen ist.

Gesamte End-to-End-Latenz der Stufe, ohne Eingabestufe Genau einmal
P50 Durchschnitt: 1.200 ms, n=3
P95 Durchschnitt: 3.000 ms, n=3
P99 Durchschnitt: 5.400 ms, n=3

Bei den Tests wurde die End-to-End-Latenz pro Stufe (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 Pipelinestufe verbringt. Er umfasst alle internen Schritte der Pipeline, z. B.:

  • Shuffling und Warteschlangen für die Verarbeitung von Nachrichten
  • 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

Aufgrund einer Einschränkung des Messwerts wird die Latenz der Eingabestufe nicht gemeldet. Daher ist sie nicht in der Gesamtsumme enthalten.

Die hier gezeigten Benchmarks stellen eine Baseline dar. Die Latenz ist sehr empfindlich gegenüber der Komplexität der Pipeline. Benutzerdefinierte UDFs, zusätzliche Transformationen und komplexe Fensterlogik können die Latenz erhöhen.

Kosten schätzen

Sie können die Basiskosten 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 Speicher 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 Modus „Genau einmal“ 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: 63 Worker × 0,1 = 6,3 Worker
  • Voraussichtliche Anzahl der Streaming Engine-Recheneinheiten pro Stunde: 58 × 0,1 = 5,8 Einheiten pro Stunde

Dieser Wert sollte nur als erste Schätzung verwendet werden. Der tatsächliche Durchsatz und die tatsächlichen 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.

Modus „Genau einmal“ 

gcloud dataflow flex-template run JOB_NAME \
  --project=PROJECT_ID \
  --template-file-gcs-location=gs://dataflow-templates-us-central1/latest/flex/Kafka_to_BigQuery_Flex \
  --enable-streaming-engine \
  --parameters \
readBootstrapServerAndTopic="KAFKA_BOOTSTRAP_ADDRESS;KAFKA_TOPIC",\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
writeMode=SINGLE_TABLE_NAME,\
outputTableSpec="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME",\
useBigQueryDLQ=true,\
outputDeadletterTable="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME_dlq",\
numStorageWriteApiStreams=400

Modus „Mindestens einmal“ 

gcloud dataflow flex-template run JOB_NAME \
  --project=PROJECT_ID \
  --template-file-gcs-location=gs://dataflow-templates-us-central1/latest/flex/Kafka_to_BigQuery_Flex \
  --enable-streaming-engine \
  --additional-experiments=streaming_mode_at_least_once \
  --parameters \
readBootstrapServerAndTopic="KAFKA_BOOTSTRAP_ADDRESS;KAFKA_TOPIC",\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
writeMode=SINGLE_TABLE_NAME,\
outputTableSpec="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME",\
useBigQueryDLQ=true,\
outputDeadletterTable="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME_dlq",\
numStorageWriteApiStreams=400,\
useStorageWriteApiAtLeastOnce=true

Ersetzen Sie Folgendes:

  • JOB_NAME: der Dataflow-Jobname
  • PROJECT_ID: die Projekt-ID
  • KAFKA_BOOTSTRAP_ADDRESS: die Bootstrap-Adresse des Apache Kafka-Clusters
  • KAFKA_TOPIC: der Name des Kafka-Themas
  • BQ_DATASET: der Name des BigQuery-Datasets
  • BQ_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_NAME \
  --project=PROJECT_ID \
  --template-file-gcs-location=gs://dataflow-templates-us-central1/latest/flex/Streaming_Data_Generator \
  --max-workers=140 \
  --parameters \
schemaLocation=SCHEMA_LOCATION,\
qps=1000000,\
sinkType=KAFKA,\
bootstrapServer=KAFKA_BOOTSTRAP_ADDRESS,\
kafkaTopic=KAFKA_TOPIC,\
outputType=JSON

Ersetzen Sie Folgendes:

  • JOB_NAME: der Dataflow-Jobname
  • PROJECT_ID: die Projekt-ID
  • SCHEMA_LOCATION: der Pfad zu einer Schemadatei in Cloud Storage
  • KAFKA_BOOTSTRAP_ADDRESS: die Bootstrap-Adresse des Apache Kafka-Clusters
  • KAFKA_TOPIC: der Name des Kafka-Themas

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