This is a tool to perform a high number of reads to a Bigtable table for

demonstration purposes. It is deployed as a Dataflow template, so it can easily

be run as a Dataflow job.

1. Set your environment variables

```

```

export TEMPLATE_PATH="gs://cloud-bigtable-dataflow-templates/generate-workload.json"

export TEMPLATE_IMAGE="gcr.io/cloud-bigtable-ecosystem/dataflow/generate-workload:latest"

export LOGS_PATH="gs://cloud-bigtable-dataflow-templates-logs/workload-generator"

```

```

gcloud dataflow flex-template build $TEMPLATE_PATH \

--image-gcr-path "$TEMPLATE_IMAGE" \

--sdk-language "JAVA" \

--flex-template-base-image JAVA11 \

--metadata-file "metadata.json" \

--jar "target/workload-generator-0.1.jar" \

--env FLEX_TEMPLATE_JAVA_MAIN_CLASS="bigtable.WorkloadGenerator" \

--gcs-log-dir="$LOGS_PATH"

```

```

GOOGLE_CLOUD_PROJECT=your-project-id

INSTANCE_ID=your-instance-id

REGION=us-central1

TABLE_ID=your-table-id

WORKLOAD_QPS=100 # Optional

```

```

mvn compile exec:java -Dexec.mainClass=WorkloadGenerator \

"-Dexec.args=--bigtableInstanceId=$INSTANCE_ID =--bigtableTableId=$TABLE_ID \

--runner=dataflow --project=$GOOGLE_CLOUD_PROJECT \

--region=$REGION" \

--workloadQPS=$WORKLOAD_QPS

```

{

"name": "Bigtable workload generator",

"description": "An Apache Beam that puts a specified read QPS to a Bigtable table.",

"parameters": [

{

"name": "bigtableInstanceId",

"label": "Bigtable instance",

"helpText": "Bigtable instance to read from.",

"regexes": [

"[A-Za-z]+[-A-Za-z0-9]*[A-Za-z0-9]+"

]

},

{

"name": "bigtableTableId",

"label": "Bigtable table",

"helpText": "Bigtable table to read from.",

"regexes": [

"[A-Za-z]+[-A-Za-z0-9]*[A-Za-z0-9]+"

]

},

{

"name": "workloadRate",

"label": "Workload Rate",

"helpText": "The QPS to put on the table (default 1000)",

"isOptional": true,

"regexes": [

"[0-9]+"

]

}

]

}

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

<modelVersion>4.0.0</modelVersion>

<groupId>com.google.cloud</groupId>

<artifactId>workload-generator</artifactId>

<version>0.1</version>

<properties>

<maven.compiler.source>8</maven.compiler.source>

<maven.compiler.target>8</maven.compiler.target>

<apache_beam.version>2.34.0</apache_beam.version>

</properties>

<!--

<parent>

<groupId>com.google.cloud.samples</groupId>

<artifactId>shared-configuration</artifactId>

<version>1.0.23</version>

</parent>

<!-- Shading plugin needed to create uber jars for Dataflow flex templates. -->

<build>

<plugins>

<plugin>

<groupId>org.apache.maven.plugins</groupId>

<artifactId>maven-shade-plugin</artifactId>

<version>3.2.4</version>

<executions>

<execution>

<phase>package</phase>

<goals>

<goal>shade</goal>

</goals>

<configuration>

<!-- Exclude signed jars which cause build issue when deploying shaded jar -->

<filters>

<filter>

<artifact>*:*</artifact>

<excludes>

<exclude>META-INF/LICENSE</exclude>

<exclude>META-INF/*.SF</exclude>

<exclude>META-INF/*.DSA</exclude>

<exclude>META-INF/*.RSA</exclude>

</excludes>

</filter>

</filters>

<transformers>

<transformer

implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer">

</transformer>

</transformers>

</configuration>

</execution>

</executions>

</plugin>

</plugins>

</build>

<dependencies>

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-runners-direct-java</artifactId>

<version>${apache_beam.version}</version>

</dependency>

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-runners-google-cloud-dataflow-java</artifactId>

<version>${apache_beam.version}</version>

</dependency>

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-sdks-java-extensions-google-cloud-platform-core

</artifactId>

<version>${apache_beam.version}</version>

</dependency>

<dependency>

<groupId>com.google.guava</groupId>

<artifactId>guava</artifactId>

<version>31.0.1-jre</version>

</dependency>

<dependency>

<groupId>com.google.cloud.bigtable</groupId>

<artifactId>bigtable-hbase-beam</artifactId>

<version>1.26.1</version>

</dependency>

<dependency>

<groupId>junit</groupId>

<artifactId>junit</artifactId>

<version>4.13.2</version>

<scope>test</scope>

</dependency>

<dependency>

<groupId>com.google.truth</groupId>

<artifactId>truth</artifactId>

<version>1.1.3</version>

<scope>test</scope>

</dependency>

</dependencies>

</project>

package bigtable;

import com.google.cloud.bigtable.beam.AbstractCloudBigtableTableDoFn;

import com.google.cloud.bigtable.beam.CloudBigtableConfiguration;

import com.google.cloud.bigtable.beam.CloudBigtableTableConfiguration;

import java.io.IOException;

import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;

import org.apache.beam.sdk.Pipeline;

import org.apache.beam.sdk.PipelineResult;

import org.apache.beam.sdk.io.GenerateSequence;

import org.apache.beam.sdk.options.Default;

import org.apache.beam.sdk.options.Description;

import org.apache.beam.sdk.options.PipelineOptions;

import org.apache.beam.sdk.options.PipelineOptionsFactory;

import org.apache.beam.sdk.transforms.ParDo;

import org.apache.hadoop.hbase.TableName;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.client.Table;

import org.joda.time.Duration;

public class WorkloadGenerator {

public static void main(String[] args) {

BigtableWorkloadOptions options =

PipelineOptionsFactory.fromArgs(args).withValidation().as(BigtableWorkloadOptions.class);

generateWorkload(options);

}

static PipelineResult generateWorkload(BigtableWorkloadOptions options) {

CloudBigtableTableConfiguration bigtableTableConfig =

new CloudBigtableTableConfiguration.Builder()

.withProjectId(options.getProject())

.withInstanceId(options.getBigtableInstanceId())

.withTableId(options.getBigtableTableId())

.build();

Pipeline p = Pipeline.create(options);

// Initiates a new pipeline every second

p.apply(GenerateSequence.from(0).withRate(options.getWorkloadRate(), new Duration(1000)))

.apply(ParDo.of(new ReadFromTableFn(bigtableTableConfig)));

System.out.println("Beginning to generate read workload.");

return p.run();

}

public static class ReadFromTableFn extends AbstractCloudBigtableTableDoFn<Long, Void> {

public ReadFromTableFn(CloudBigtableConfiguration config) {

super(config);

System.out.println("Connected to table.");

}

@ProcessElement

public void processElement(PipelineOptions po) throws IOException {

BigtableWorkloadOptions options = po.as(BigtableWorkloadOptions.class);

Scan scan = new Scan();

Table table = getConnection().getTable(TableName.valueOf(options.getBigtableTableId()));

table.getScanner(scan);

}

}

public interface BigtableWorkloadOptions extends DataflowPipelineOptions {

@Description("The Bigtable instance ID")

@Default.String("bigtable-instance")

String getBigtableInstanceId();

void setBigtableInstanceId(String bigtableInstanceId);

@Description("The Bigtable table ID in the instance.")

@Default.String("bigtable-table")

String getBigtableTableId();

void setBigtableTableId(String bigtableTableId);

@Description("The QPS for the workload to produce.")

@Default.Integer(1000)

Integer getWorkloadRate();

void setWorkloadRate(Integer workloadRate);

}

}