Streaming Data Generator template

The Streaming Data Generator template generates synthetic records or messages and sends them to a destination sink. You can configure the record schema and the rate at which records are generated.

The template supports the following sinks:

  • Apache Kafka topic
  • BigQuery table
  • Cloud Storage bucket
  • Java Database Connectivity (JDBC) endpoint
  • Pub/Sub topic
  • Spanner table

Following are a set of few possible use cases:

  • Simulate large-scale real-time event publishing to a Pub/Sub topic to measure and determine the number and size of consumers required to process published events.
  • Generate synthetic data to evaluate performance benchmarks or serve as a proof of concept.
  • Validate an end-to-end pipeline. For example, send records to a Kafka topic, which are then read by a downstream consumer.

Pipeline requirements

Define the record schema

The template provides a predefined schema for the generated data. To use this schema, set the schemaTemplate template parameter to GAME_EVENT.

Alternatively, you can provide your own data schema as follows:

  1. Create a schema file that contains a JSON template for the generated data. This template uses the JSON Data Generator library, which supports various functions for randomizing the data. For example:

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

    For more information, see the json-data-generator documentation.

  2. Upload the schema file to a Cloud Storage bucket.
  3. Set the schemaLocation template parameter to the Cloud Storage URI of the template file.

Specify the output format

By default, the template produces JSON data. For some destinations, the template also supports Avro or Parquet formats:

  • Avro: Supported for Cloud Storage, Apache Kafka, and Pub/Sub
  • Parquet: Supported for Cloud Storage.

To output Avro or Parquet format, do the following:

  1. Set the outputType template parameter to AVRO for Avro format, or PARQUET for Parquet format.
  2. Create an Avro schema file.
  3. Upload the schema file to Cloud Storage.
  4. Set the avroSchemaLocation template parameter to the Cloud Storage URI of the schema file.

Specify the destination sink

The following sections describe how to configure the template for each type of sink.

Apache Kafka topic

To write to a Kafka topic, set the following template parameters:

  • sinkType: KAFKA.
  • bootstrapServer: The bootstrap address of the Kafka cluster.
  • kafkaTopic: The Kafka topic to write to.

If you are writing to a Google Cloud Managed Service for Apache Kafka cluster, grant the worker service account the Managed Kafka Client (roles/managedkafka.client) role.

BigQuery table

To write to a BigQuery table, set the following template parameters:

  • sinkType: BIGQUERY.
  • outputTableSpec: The BigQuery table to write to. Format this parameter as follows: PROJECT_ID:DATASET.TABLE.

The following parameters are optional:

  • outputDeadletterTable: The name of the table where the pipeline writes failed records. If not specified, the pipeline creates a table named OUTPUT_TABLE_error_records, where OUTPUT_TABLE is the name of the output table.

  • writeDisposition: Specifies how to write to an existing table. The following values are supported:

    • WRITE_APPEND. Append rows to the existing table.
    • WRITE_TRUNCATE. Truncate the existing rows.
    • WRITE_EMPTY. Write only if the table is empty. If the table already has data, the job fails.

    The default value is WRITE_APPEND.

Grant the worker service account the BigQuery Data Editor (roles/bigquery.dataEditor) role.

Cloud Storage

To write to a Cloud Storage bucket, set the following template parameters:

  • sinkType: GCS.
  • outputDirectory: The path of the Cloud Storage folder to write to.

The following parameters are optional:

  • numShards: The maximum number of shards. A higher value can enable higher throughput, but potentially higher data aggregation costs. If the value is 0, Dataflow selects the number of shards. The default value is 0.
  • outputFilenamePrefix: The filename prefix. The default value is output-.
  • windowDuration: The interval at which the pipeline writes files to Cloud Storage. Allowed formats are Ns (seconds), Nm (minutes), and Nh (hours). The default value is 1m (1 minute).

Grant the worker service account the Storage Object Admin (roles/storage.objectAdmin) role.

JDBC endpoint

To write to a JDBC endpoint, set the following template parameters:

  • sinkType: JDBC.
  • driverClassName: The JDBC driver class to use. Example: com.mysql.jdbc.Driver.
  • connectionUrl: The connection string to connect to the JDBC source.

  • statement: The INSERT INTO SQL statement used to write to the database. The statement must specify which table columns to write, with placeholder '?' characters for the VALUES clause. The pipeline replaces the placeholders with the corresponding field values from the JSON data.

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

The following parameters are optional:

  • username: The username for the JDBC connection.
  • password: The password for the JDBC connection.
  • connectionProperties: A properties string for the JDBC connection. Example: unicode=true;characterEncoding=UTF-8.

Pub/Sub topic

To write to a Pub/Sub topic, set the following template parameters:

  • sinkType: PUBSUB.
  • topic: The Pub/Sub topic to write to.

Grant the worker service account the Pub/Sub Publisher (roles/pubsub.publisher) role.

Spanner table

To write to a Spanner table, set the following template parameters:

  • sinkType: SPANNER.
  • projectId: The ID of the project that contains the Spanner table.
  • spannerInstanceName: The name of the Spanner instance.
  • spannerDatabaseName: The name of the Spanner database.
  • spannerTableName: The name of the Spanner table.

The following parameters are optional:

  • maxNumMutations: The maximum number of mutated cells per batch.
  • maxNumRows: The maximum number of mutated rows per batch.
  • batchSizeBytes: The maximum number of bytes mutated per batch.
  • commitDeadlineSeconds: The deadline for the commit API call, in seconds.

Grant the worker service account the Cloud Spanner Database User (roles/spanner.databaseUser) role.

Template parameters

Required parameters

  • qps: Indicates rate of messages per second to be published to Pub/Sub.

Optional parameters

  • schemaTemplate: Pre-existing schema template to use. The value must be one of: [GAME_EVENT].
  • schemaLocation: Cloud Storage path of schema location. For example, gs://<bucket-name>/prefix.
  • topic: The name of the topic to which the pipeline should publish data. For example, projects/<project-id>/topics/<topic-name>.
  • messagesLimit: Indicates maximum number of output messages to be generated. 0 means unlimited. Defaults to: 0.
  • outputType: The message Output type. Default is JSON.
  • avroSchemaLocation: Cloud Storage path of Avro schema location. Mandatory when output type is AVRO or PARQUET. For example, gs://your-bucket/your-path/schema.avsc.
  • sinkType: The message Sink type. Default is PUBSUB.
  • outputTableSpec: Output BigQuery table. Mandatory when sinkType is BIGQUERY For example, <project>:<dataset>.<table_name>.
  • writeDisposition: BigQuery WriteDisposition. For example, WRITE_APPEND, WRITE_EMPTY or WRITE_TRUNCATE. Defaults to: WRITE_APPEND.
  • outputDeadletterTable: Messages failed to reach the output table for all kind of reasons (e.g., mismatched schema, malformed json) are written to this table. If it doesn't exist, it will be created during pipeline execution. For example, your-project-id:your-dataset.your-table-name.
  • windowDuration: The window duration/size in which data will be written to Cloud Storage. Allowed formats are: Ns (for seconds, example: 5s), Nm (for minutes, example: 12m), Nh (for hours, example: 2h). For example, 1m. Defaults to: 1m.
  • outputDirectory: The path and filename prefix for writing output files. Must end with a slash. DateTime formatting is used to parse directory path for date & time formatters. For example, gs://your-bucket/your-path/.
  • outputFilenamePrefix: The prefix to place on each windowed file. For example, output-. Defaults to: output-.
  • numShards: The maximum number of output shards produced when writing. A higher number of shards means higher throughput for writing to Cloud Storage, but potentially higher data aggregation cost across shards when processing output Cloud Storage files. Default value is decided by Dataflow.
  • driverClassName: JDBC driver class name to use. For example, com.mysql.jdbc.Driver.
  • connectionUrl: Url connection string to connect to the JDBC source. For example, jdbc:mysql://some-host:3306/sampledb.
  • username: User name to be used for the JDBC connection.
  • password: Password to be used for the JDBC connection.
  • connectionProperties: Properties string to use for the JDBC connection. Format of the string must be [propertyName=property;]*. For example, unicode=true;characterEncoding=UTF-8.
  • statement: SQL statement which will be executed to write to the database. The statement must specify the column names of the table in any order. Only the values of the specified column names will be read from the json and added to the statement. For example, INSERT INTO tableName (column1, column2) VALUES (?,?).
  • projectId: GCP Project Id of where the Spanner table lives.
  • spannerInstanceName: Cloud Spanner instance name.
  • spannerDatabaseName: Cloud Spanner database name.
  • spannerTableName: Cloud Spanner table name.
  • maxNumMutations: Specifies the cell mutation limit (maximum number of mutated cells per batch). Default value is 5000.
  • maxNumRows: Specifies the row mutation limit (maximum number of mutated rows per batch). Default value is 1000.
  • batchSizeBytes: Specifies the batch size limit (max number of bytes mutated per batch). Default value is 1MB.
  • commitDeadlineSeconds: Specifies the deadline in seconds for the Commit API call.
  • bootstrapServer: Kafka Bootstrap Server For example, localhost:9092.
  • kafkaTopic: Kafka topic to write to. For example, topic.

Run the template

Console

  1. Go to the Dataflow Create job from template page.
    2. Go to Create job from template
  2. In the Job name field, enter a unique job name.
  3. Optional: For Regional endpoint, select a value from the drop-down menu. The default region is us-central1.

    For a list of regions where you can run a Dataflow job, see Dataflow locations.

  4. From the Dataflow template drop-down menu, select the Streaming Data Generator template.
  5. In the provided parameter fields, enter your parameter values.
  6. Click Run job.

gcloud

In your shell or terminal, run the template:

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

Replace the following:

  • PROJECT_ID: the Google Cloud project ID where you want to run the Dataflow job
  • REGION_NAME: the region where you want to deploy your Dataflow job—for example, us-central1
  • JOB_NAME: a unique job name of your choice
  • VERSION: the version of the template that you want to use

    You can use the following values:

  • SCHEMA_LOCATION: the path to schema file in Cloud Storage. For example: gs://mybucket/filename.json.
  • QPS: the number of messages to be published per second
  • PUBSUB_TOPIC: the output Pub/Sub topic. For example: projects/my-project-id/topics/my-topic-id.

API

To run the template using the REST API, send an HTTP POST request. For more information on the API and its authorization scopes, see 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/",
   }
}

Replace the following:

  • PROJECT_ID: the Google Cloud project ID where you want to run the Dataflow job
  • LOCATION: the region where you want to deploy your Dataflow job—for example, us-central1
  • JOB_NAME: a unique job name of your choice
  • VERSION: the version of the template that you want to use

    You can use the following values:

  • SCHEMA_LOCATION: the path to schema file in Cloud Storage. For example: gs://mybucket/filename.json.
  • QPS: the number of messages to be published per second
  • PUBSUB_TOPIC: the output Pub/Sub topic. For example: projects/my-project-id/topics/my-topic-id.

What's next