Vorlage für Streaming Data Generator

Mit der Vorlage für Streaming Data Generator werden synthetische Datensätze oder Nachrichten generiert und an eine Zielsenke gesendet. Sie können das Datensatzschema und die Rate konfigurieren, mit der Datensätze generiert werden.

Die Vorlage unterstützt die folgenden Senken:

  • Apache Kafka-Thema
  • BigQuery-Tabelle
  • Cloud Storage-Bucket
  • JDBC-Endpunkt (Java Database Connectivity)
  • Pub/Sub-Thema
  • Cloud Spanner-Tabelle

Im Folgenden sind einige mögliche Anwendungsfälle aufgeführt:

  • Simulieren der Veröffentlichung von Echtzeitereignissen in großem Umfang in einem Pub/Sub-Thema, um die Anzahl und Größe der Nutzer zu ermitteln und zu bestimmen, die zur Verarbeitung veröffentlichter Ereignisse erforderlich sind.
  • Generieren Sie synthetische Daten, um Leistungs-Benchmarks zu bewerten oder um diese als Proof of Concept zu nutzen.
  • End-to-End-Pipeline validieren Sie können beispielsweise Datensätze an ein Kafka-Thema senden, die dann von einem Downstream-Consumer gelesen werden.

Pipelineanforderungen

Datensatzschema definieren

Die Vorlage enthält ein vordefiniertes Schema für die generierten Daten. Wenn Sie dieses Schema verwenden möchten, legen Sie den Vorlagenparameter schemaTemplate auf GAME_EVENT fest.

Alternativ können Sie Ihr eigenes Datenschema so angeben:

  1. Erstellen Sie eine Schemadatei, die eine JSON-Vorlage für die generierten Daten enthält. In dieser Vorlage wird die Bibliothek JSON Data Generator verwendet, die verschiedene Funktionen zum Randomisieren der Daten unterstützt. Beispiel:

    {
  "id": {{integer(0,1000)}},
  "name": "{{uuid()}}",
  "isInStock": {{bool()}}
}

    Weitere Informationen finden Sie in der Dokumentation zu json-data-generator.

  2. Laden Sie die Schemadatei in einen Cloud Storage-Bucket hoch.
  3. Legen Sie für den Vorlagenparameter schemaLocation den Cloud Storage-URI der Vorlagendatei fest.

Ausgabeformat angeben

Standardmäßig werden mit der Vorlage JSON-Daten generiert. Für einige Ziele unterstützt die Vorlage auch die Formate Avro oder Parquet:

  • Avro: Wird für Cloud Storage, Apache Kafka und Pub/Sub unterstützt.
  • Parquet: Wird für Cloud Storage unterstützt.

So geben Sie das Avro- oder Parquet-Format aus:

  1. Legen Sie den Vorlagenparameter outputType für das Avro-Format auf AVRO und für das Parquet-Format auf PARQUET fest.
  2. Erstellen Sie eine Avro-Schemadatei.
  3. Laden Sie die Schemadatei in Cloud Storage hoch.
  4. Legen Sie für den Vorlagenparameter avroSchemaLocation den Cloud Storage-URI der Schemadatei fest.

Zielsenke angeben

In den folgenden Abschnitten wird beschrieben, wie Sie die Vorlage für die einzelnen Senkentypen konfigurieren.

Apache Kafka-Thema

Wenn Sie in ein Kafka-Thema schreiben möchten, legen Sie die folgenden Vorlagenparameter fest:

  • sinkType: KAFKA.
  • bootstrapServer: Die Bootstrap-Adresse des Kafka-Clusters.
  • kafkaTopic: Das Kafka-Thema, in das geschrieben werden soll.

Wenn Sie in einen Google Cloud Managed Service for Apache Kafka-Cluster schreiben, gewähren Sie dem Worker-Dienstkonto die Rolle „Managed Kafka Client“ (roles/managedkafka.client).

BigQuery-Tabelle

Wenn Sie in eine BigQuery-Tabelle schreiben möchten, legen Sie die folgenden Vorlagenparameter fest:

  • sinkType: BIGQUERY.
  • outputTableSpec: Die BigQuery-Tabelle, in die geschrieben werden soll. Formatieren Sie diesen Parameter so: PROJECT_ID:DATASET.TABLE.

Folgende Parameter sind optional:

  • outputDeadletterTable: Der Name der Tabelle, in die die Pipeline fehlgeschlagene Datensätze schreibt. Wenn nicht angegeben, wird von der Pipeline eine Tabelle mit dem Namen OUTPUT_TABLE_error_records erstellt, wobei OUTPUT_TABLE der Name der Ausgabetabelle ist.

  • writeDisposition: Gibt an, wie in eine vorhandene Tabelle geschrieben werden soll. Folgende Werte werden unterstützt:

    • WRITE_APPEND. Zeilen an die vorhandene Tabelle anhängen.
    • WRITE_TRUNCATE. Kürzen Sie die vorhandenen Zeilen.
    • WRITE_EMPTY. Nur schreiben, wenn die Tabelle leer ist. Wenn die Tabelle bereits Daten enthält, schlägt der Job fehl.

    Der Standardwert ist WRITE_APPEND.

Weisen Sie dem Worker-Dienstkonto die Rolle „BigQuery-Datenbearbeiter“ (roles/bigquery.dataEditor) zu.

Cloud Storage

Wenn Sie in einen Cloud Storage-Bucket schreiben möchten, legen Sie die folgenden Vorlagenparameter fest:

  • sinkType: GCS.
  • outputDirectory: Der Pfad des Cloud Storage-Ordners, in den geschrieben werden soll.

Folgende Parameter sind optional:

  • numShards: Die maximale Anzahl von Shards. Ein höherer Wert kann einen höheren Durchsatz ermöglichen, aber möglicherweise auch höhere Kosten für die Datenaggregation verursachen. Wenn der Wert 0 ist, wählt Dataflow die Anzahl der Shards aus. Der Standardwert ist 0.
  • outputFilenamePrefix: Das Präfix des Dateinamens. Der Standardwert ist output-.
  • windowDuration: Das Intervall, in dem die Pipeline Dateien in Cloud Storage schreibt. Zulässige Formate sind Ns (Sekunden), Nm (Minuten) und Nh (Stunden). Der Standardwert ist 1m (1 Minute).

Weisen Sie dem Worker-Dienstkonto die Rolle „Storage-Objekt-Administrator“ (roles/storage.objectAdmin) zu.

JDBC-Endpunkt

Wenn Sie Daten in einen JDBC-Endpunkt schreiben möchten, legen Sie die folgenden Vorlagenparameter fest:

  • sinkType: JDBC.
  • driverClassName: Die zu verwendende JDBC-Treiberklasse. Beispiel: com.mysql.jdbc.Driver
  • connectionUrl: Der Verbindungsstring für die Verbindung zur JDBC-Quelle.

  • statement: Die INSERT INTO-SQL-Anweisung, die zum Schreiben in die Datenbank verwendet wird. In der Anweisung muss angegeben werden, in welche Tabellenspalten geschrieben werden soll. Verwenden Sie dazu Platzhalterzeichen '?' für die VALUES-Klausel. Die Pipeline ersetzt die Platzhalter durch die entsprechenden Feldwerte aus den JSON-Daten.

    Beispiel: INSERT INTO tableName (column1, column2) VALUES (?,?).

Folgende Parameter sind optional:

  • username: Der Nutzername für die JDBC-Verbindung.
  • password: Das Passwort für die JDBC-Verbindung.
  • connectionProperties: Ein Attributstring für die JDBC-Verbindung. Beispiel: unicode=true;characterEncoding=UTF-8.

Pub/Sub-Thema

Wenn Sie in ein Pub/Sub-Thema schreiben möchten, legen Sie die folgenden Vorlagenparameter fest:

  • sinkType: PUBSUB.
  • topic: Das Pub/Sub-Thema, in das geschrieben werden soll.

Weisen Sie dem Worker-Dienstkonto die Rolle „Pub/Sub-Publisher“ (roles/pubsub.publisher) zu.

Cloud Spanner-Tabelle

Wenn Sie in eine Spanner-Tabelle schreiben möchten, legen Sie die folgenden Vorlagenparameter fest:

  • sinkType: SPANNER.
  • projectId: Die ID des Projekts, das die Cloud Spanner-Tabelle enthält.
  • spannerInstanceName: Der Name der Spanner-Instanz.
  • spannerDatabaseName: Der Name der Spanner-Datenbank.
  • spannerTableName: Der Name der Spanner-Tabelle.

Folgende Parameter sind optional:

  • maxNumMutations: Die maximale Anzahl der mutierten Zellen pro Batch.
  • maxNumRows: Die maximale Anzahl der geänderten Zeilen pro Batch.
  • batchSizeBytes: Die maximale Anzahl von Byte, die pro Batch mutiert werden.
  • commitDeadlineSeconds: Die Frist für den Commit-API-Aufruf in Sekunden.

Weisen Sie dem Worker-Dienstkonto die Rolle „Cloud Spanner-Datenbanknutzer“ (roles/spanner.databaseUser) zu.

Vorlagenparameter

Erforderliche Parameter

  • qps: Gibt die Rate der Nachrichten pro Sekunde an, die in Pub/Sub veröffentlicht werden sollen.

Optionale Parameter

  • schemaTemplate: Vorhandene Schemavorlage, die verwendet werden soll. Der Wert muss einer der folgenden sein: [GAME_EVENT].
  • schemaLocation: Cloud Storage-Pfad des Schemaspeicherorts. Beispiel: gs://<bucket-name>/prefix.
  • topic: Der Name des Themas, in dem die Pipeline Daten veröffentlichen soll. Beispiel: projects/<project-id>/topics/<topic-name>.
  • messagesLimit: Gibt die maximale Anzahl der zu generierenden Ausgabenachrichten an. 0 bedeutet unbegrenzt. Die Standardeinstellung ist 0.
  • outputType: Der Ausgabetyp der Nachricht. Der Standardwert ist JSON.
  • avroSchemaLocation: Cloud Storage-Pfad zum Speicherort des Avro-Schemas. Erforderlich, wenn der Ausgabetyp AVRO oder PARQUET ist. Beispiel: gs://your-bucket/your-path/schema.avsc.
  • sinkType: Der Typ der Nachrichtensenke. Der Standardwert ist PUBSUB.
  • outputTableSpec: BigQuery-Tabelle. Obligatorisch, wenn „sinkType“ BIGQUERY ist, z. B. <project>:<dataset>.<table_name>.
  • writeDisposition: BigQuery-WriteDisposition. Beispiele: WRITE_APPEND, WRITE_EMPTY oder WRITE_TRUNCATE. Die Standardeinstellung ist WRITE_APPEND.
  • outputDeadletterTable: Nachrichten, die die Ausgabetabelle aus irgendeinem Grund nicht erreicht haben (z.B. nicht übereinstimmendes Schema, fehlerhaft formatierte JSON-Datei), werden in diese Tabelle geschrieben. Wenn sie nicht vorhanden ist, wird sie während der Pipelineausführung erstellt. Beispiel: your-project-id:your-dataset.your-table-name.
  • windowDuration: Die Fensterdauer/Größe, in der Daten in Cloud Storage geschrieben werden. Zulässige Formate sind: Ns (für Sekunden, Beispiel: 5s), Nm (für Minuten, Beispiel: 12m), Nh (für Stunden, Beispiel: 2h). Beispiel: 1m. Die Standardeinstellung ist „1m“.
  • outputDirectory: Das Pfad- und Dateinamenpräfix zum Schreiben von Ausgabedateien. Muss mit einem Schrägstrich enden. Die DateTime-Formatierung wird verwendet, um den Verzeichnispfad für Datums- und Uhrzeitformatierer zu parsen. Beispiel: gs://your-bucket/your-path/.
  • outputFilenamePrefix: Das Präfix für die Namen der einzelnen Dateien im Fenstermodus. Beispiel: output-. Die Standardeinstellung ist „output-“.
  • numShards: Die maximale Anzahl von Ausgabe-Shards, die beim Schreiben erzeugt werden. Eine höhere Anzahl von Shards erhöht den Durchsatz für das Schreiben in Cloud Storage, aber möglicherweise auch höhere Kosten für die Datenaggregation über Shards bei der Verarbeitung von Cloud Storage-Ausgabedateien. Der Standardwert wird von Dataflow festgelegt.
  • driverClassName: Der Name der zu verwendenden JDBC-Treiberklasse. Beispiel: com.mysql.jdbc.Driver.
  • connectionUrl: URL-Verbindungs-String für die Verbindung zur JDBC-Quelle. Beispiel: jdbc:mysql://some-host:3306/sampledb.
  • Nutzername: Der Nutzername, der für die JDBC-Verbindung verwendet werden soll.
  • Passwort: Das Passwort für die JDBC-Verbindung.
  • connectionProperties: Attributstring für die JDBC-Verbindung. Format des Strings muss [propertyName=property;]* sein. Beispiel: unicode=true;characterEncoding=UTF-8.
  • statement: SQL-Anweisung, die zum Schreiben in die Datenbank ausgeführt wird. Die Anweisung muss die Spaltennamen der Tabelle in beliebiger Reihenfolge angeben. Nur die Werte der angegebenen Spaltennamen werden aus der JSON-Datei gelesen und der Anweisung hinzugefügt. Beispiel: INSERT INTO tableName (column1, column2) VALUES (?,?).
  • projectId: GCP-Projekt-ID des Projekts, in dem sich die Spanner-Tabelle befindet.
  • spannerInstanceName: Name der Cloud Spanner-Instanz.
  • spannerDatabaseName: Name der Cloud Spanner-Datenbank.
  • spannerTableName: Der Name der Cloud Spanner-Tabelle.
  • maxNumMutations: Gibt das Limit für Zellmutationen an (maximale Anzahl von mutierten Zellen pro Batch). Der Standardwert ist 5.000.
  • maxNumRows: Gibt das Limit für Zeilenmutationen an (maximale Anzahl mutierter Zeilen pro Batch). Der Standardwert ist 1.000.
  • batchSizeBytes: Gibt das Batchgrößenlimit an (maximale Anzahl von Byte, die pro Batch geändert werden). Der Standardwert ist 1 MB.
  • commitDeadlineSeconds: Gibt die Frist in Sekunden für den Commit-API-Aufruf an.
  • bootstrapServer: Kafka-Bootstrap-Server, z. B. localhost:9092.
  • kafkaTopic: Kafka-Thema, in das geschrieben werden soll. Beispiel: topic.

Führen Sie die Vorlage aus.

Console

  1. Rufen Sie die Dataflow-Seite Job aus Vorlage erstellen auf.
    2. Zur Seite "Job aus Vorlage erstellen“
  2. Geben Sie im Feld Jobname einen eindeutigen Jobnamen ein.
  3. Optional: Wählen Sie für Regionaler Endpunkt einen Wert aus dem Drop-down-Menü aus. Die Standardregion ist us-central1.

    Eine Liste der Regionen, in denen Sie einen Dataflow-Job ausführen können, finden Sie unter Dataflow-Standorte.

  4. Wählen Sie im Drop-down-Menü Dataflow-Vorlage die Vorlage Streaming Data Generator aus.
  5. Geben Sie Ihre Parameterwerte in die Parameterfelder ein.
  6. Klicken Sie auf Job ausführen.

gcloud

Führen Sie die Vorlage in der Shell oder im Terminal aus:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/ \
    --parameters \
schemaLocation=SCHEMA_LOCATION,\
qps=QPS,\
topic=PUBSUB_TOPIC

Ersetzen Sie Folgendes:

  • PROJECT_ID: Die Google Cloud Projekt-ID, in der Sie den Dataflow-Job ausführen möchten
  • REGION_NAME: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
  • JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
  • VERSION: Die Version der Vorlage, die Sie verwenden möchten

    Sie können die folgenden Werte verwenden:

    • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
    • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
  • SCHEMA_LOCATION: der Pfad zur Schemadatei in Cloud Storage, z. B. gs://mybucket/filename.json
  • QPS: die Anzahl der Nachrichten, die pro Sekunde zu veröffentlicht werden sollen
  • PUBSUB_TOPIC: das Pub/Sub-Ausgabethema Beispiel: projects/my-project-id/topics/my-topic-id.

API

Senden Sie eine HTTP-POST-Anfrage, um die Vorlage mithilfe der REST API auszuführen. Weitere Informationen zur API und ihren Autorisierungsbereichen finden Sie unter projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "schemaLocation": "SCHEMA_LOCATION",
          "qps": "QPS",
          "topic": "PUBSUB_TOPIC"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/",
   }
}

Ersetzen Sie Folgendes:

  • PROJECT_ID: Die Google Cloud Projekt-ID, in der Sie den Dataflow-Job ausführen möchten
  • LOCATION: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
  • JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
  • VERSION: Die Version der Vorlage, die Sie verwenden möchten

    Sie können die folgenden Werte verwenden:

    • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
    • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
  • SCHEMA_LOCATION: der Pfad zur Schemadatei in Cloud Storage, z. B. gs://mybucket/filename.json
  • QPS: die Anzahl der Nachrichten, die pro Sekunde zu veröffentlicht werden sollen
  • PUBSUB_TOPIC: das Pub/Sub-Ausgabethema Beispiel: projects/my-project-id/topics/my-topic-id.

Nächste Schritte