Google-provided batch templates

Google provides a set of open-source Dataflow templates. For general information about templates, see the Overview page. For a list of all Google-provided templates, see the Get started with Google-provided templates page.

This page documents batch templates:

• Cloud Bigtable to Cloud Storage Avro
• Cloud Bigtable to Cloud Storage SequenceFile
• Datastore to Cloud Storage Text
• Cloud Spanner to Cloud Storage Avro
• Cloud Spanner to Cloud Storage Text
• Cloud Storage Avro to Cloud Bigtable
• Cloud Storage Avro to Cloud Spanner
• Cloud Storage SequenceFile to Cloud Bigtable
• Cloud Storage Text to BigQuery
• Cloud Storage Text to Datastore
• Cloud Storage Text to Pub/Sub (Batch)
• Cloud Storage Text to Cloud Spanner
• Java Database Connectivity (JDBC) to BigQuery

Cloud Bigtable to Cloud Storage Avro

The Cloud Bigtable to Cloud Storage Avro template is a pipeline that reads data from a Cloud Bigtable table and writes it to a Cloud Storage bucket in Avro format. You can use the template to move data from Cloud Bigtable to Cloud Storage.

Requirements for this pipeline:

• The Cloud Bigtable table must exist.
• The output Cloud Storage bucket must exist prior to running the pipeline.

Template parameters

Parameter	Description
bigtableProjectId	The ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
bigtableInstanceId	The ID of the Cloud Bigtable instance that contains the table.
bigtableTableId	The ID of the Cloud Bigtable table to export.
outputDirectory	Cloud Storage path where data should be written. For example, gs://mybucket/somefolder.
filenamePrefix	The prefix of the Avro file name. For example, output-.

Running the Cloud Bigtable to Cloud Storage Avro file template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field. Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace [PROJECT_ID] with the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
• Replace [INSTANCE_ID] with the ID of the Cloud Bigtable instance that contains the table.
• Replace [TABLE_ID] with the ID of the Cloud Bigtable table to export.
• Replace [OUTPUT_DIRECTORY] with Cloud Storage path where data should be written. For example, gs://mybucket/somefolder.
• Replace [FILENAME_PREFIX] with the prefix of the Avro file name. For example, output-.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/ \
    --parameters bigtableProjectId=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],outputDirectory=[OUTPUT_DIRECTORY],filenamePrefix=[FILENAME_PREFIX]

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

To Run this template with a REST API request, send an HTTP POST request with your project ID. This request requires authorization.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace [PROJECT_ID] with the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
• Replace [INSTANCE_ID] with the ID of the Cloud Bigtable instance that contains the table.
• Replace [TABLE_ID] with the ID of the Cloud Bigtable table to export.
• Replace [OUTPUT_DIRECTORY] with Cloud Storage path where data should be written. For example, gs://mybucket/somefolder.
• Replace [FILENAME_PREFIX] with the prefix of the Avro file name. For example, output-.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProjectId": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "outputDirectory": "[OUTPUT_DIRECTORY]",
       "filenamePrefix": "[FILENAME_PREFIX]",
   },
   "environment": { "zone": "us-central1-f" }
}

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.bigtable;

import com.google.bigtable.v2.Cell;
import com.google.bigtable.v2.Column;
import com.google.bigtable.v2.Family;
import com.google.bigtable.v2.Row;
import com.google.cloud.teleport.util.DualInputNestedValueProvider;
import com.google.cloud.teleport.util.DualInputNestedValueProvider.TranslatorInput;
import com.google.protobuf.ByteOutput;
import com.google.protobuf.ByteString;
import com.google.protobuf.UnsafeByteOperations;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
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.AvroIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
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.options.ValueProvider;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.SimpleFunction;

/**
 * Dataflow pipeline that exports data from a Cloud Bigtable table to Avro files in GCS. Currently,
 * filtering on Cloud Bigtable table is not supported.
 */
public class BigtableToAvro {

  /** Options for the export pipeline. */
  public interface Options extends PipelineOptions {
    @Description("The project that contains the table to export.")
    ValueProvider<String> getBigtableProjectId();

    @SuppressWarnings("unused")
    void setBigtableProjectId(ValueProvider<String> projectId);

    @Description("The Bigtable instance id that contains the table to export.")
    ValueProvider<String> getBigtableInstanceId();

    @SuppressWarnings("unused")
    void setBigtableInstanceId(ValueProvider<String> instanceId);

    @Description("The Bigtable table id to export.")
    ValueProvider<String> getBigtableTableId();

    @SuppressWarnings("unused")
    void setBigtableTableId(ValueProvider<String> tableId);

    @Description("The output location to write to (e.g. gs://mybucket/somefolder/)")
    ValueProvider<String> getOutputDirectory();

    @SuppressWarnings("unused")
    void setOutputDirectory(ValueProvider<String> outputDirectory);

    @Description("The prefix for each exported file in outputDirectory")
    ValueProvider<String> getFilenamePrefix();

    @SuppressWarnings("unused")
    void setFilenamePrefix(ValueProvider<String> filenamePrefix);
  }

  /**
   * Runs a pipeline to export data from a Cloud Bigtable table to Avro files in GCS.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    PipelineResult result = run(options);

    // Wait for pipeline to finish only if it is not constructing a template.
    if (options.as(DataflowPipelineOptions.class).getTemplateLocation() == null) {
      result.waitUntilFinish();
    }
  }

  public static PipelineResult run(Options options) {
    Pipeline pipeline = Pipeline.create(options);

    BigtableIO.Read read =
        BigtableIO.read()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId());

    // Do not validate input fields if it is running as a template.
    if (options.as(DataflowPipelineOptions.class).getTemplateLocation() != null) {
      read = read.withoutValidation();
    }

    ValueProvider<String> filePathPrefix = DualInputNestedValueProvider.of(
        options.getOutputDirectory(),
        options.getFilenamePrefix(),
        new SerializableFunction<TranslatorInput<String, String>, String>() {
          @Override
          public String apply(TranslatorInput<String, String> input) {
            return new StringBuilder(input.getX()).append(input.getY()).toString();
          }
        });

    pipeline
        .apply("Read from Bigtable", read)
        .apply("Transform to Avro", MapElements.via(new BigtableToAvroFn()))
        .apply(
            "Write to Avro in GCS",
            AvroIO.write(BigtableRow.class).to(filePathPrefix).withSuffix(".avro"));

    return pipeline.run();
  }

  /** Translates Bigtable {@link Row} to Avro {@link BigtableRow}. */
  static class BigtableToAvroFn extends SimpleFunction<Row, BigtableRow> {
    @Override
    public BigtableRow apply(Row row) {
      ByteBuffer key = ByteBuffer.wrap(toByteArray(row.getKey()));
      List<BigtableCell> cells = new ArrayList<>();
      for (Family family : row.getFamiliesList()) {
        String familyName = family.getName();
        for (Column column : family.getColumnsList()) {
          ByteBuffer qualifier = ByteBuffer.wrap(toByteArray(column.getQualifier()));
          for (Cell cell : column.getCellsList()) {
            long timestamp = cell.getTimestampMicros();
            ByteBuffer value = ByteBuffer.wrap(toByteArray(cell.getValue()));
            cells.add(new BigtableCell(familyName, qualifier, timestamp, value));
          }
        }
      }
      return new BigtableRow(key, cells);
    }
  }

  /**
   * Extracts the byte array from the given {@link ByteString} without copy.
   *
   * @param byteString A {@link ByteString} from which to extract the array.
   * @return an array of byte.
   */
  protected static byte[] toByteArray(final ByteString byteString) {
    try {
      ZeroCopyByteOutput byteOutput = new ZeroCopyByteOutput();
      UnsafeByteOperations.unsafeWriteTo(byteString, byteOutput);
      return byteOutput.bytes;
    } catch (IOException e) {
      return byteString.toByteArray();
    }
  }

  private static final class ZeroCopyByteOutput extends ByteOutput {
    private byte[] bytes;

    @Override
    public void writeLazy(byte[] value, int offset, int length) {
      if (offset != 0 || length != value.length) {
        throw new UnsupportedOperationException();
      }
      bytes = value;
    }

    @Override
    public void write(byte value) {
      throw new UnsupportedOperationException();
    }

    @Override
    public void write(byte[] value, int offset, int length) {
      throw new UnsupportedOperationException();
    }

    @Override
    public void write(ByteBuffer value) {
      throw new UnsupportedOperationException();
    }

    @Override
    public void writeLazy(ByteBuffer value) {
      throw new UnsupportedOperationException();
    }
  }
}

Cloud Bigtable to Cloud Storage SequenceFile

The Cloud Bigtable to Cloud Storage SequenceFile template is a pipeline that reads data from a Cloud Bigtable table and writes the data to a Cloud Storage bucket in SequenceFile format. You can use the template to copy data from Cloud Bigtable to Cloud Storage.

Requirements for this pipeline:

• The Cloud Bigtable table must exist.
• The output Cloud Storage bucket must exist prior to running the pipeline.

Template parameters

Parameter	Description
bigtableProject	The ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
bigtableInstanceId	The ID of the Cloud Bigtable instance that contains the table.
bigtableTableId	The ID of the Cloud Bigtable table to export.
bigtableAppProfileId	The ID of the Cloud Bigtable application profile to be used for the export. If you do not specify an app profile, Cloud Bigtable uses the instance's default app profile.
destinationPath	Cloud Storage path where data should be written. For example, gs://mybucket/somefolder.
filenamePrefix	The prefix of the SequenceFile file name. For example, output-.

Running the Cloud Bigtable to Cloud Storage SequenceFile template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field. Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace [PROJECT_ID] with the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
• Replace [INSTANCE_ID] with the ID of the Cloud Bigtable instance that contains the table.
• Replace [TABLE_ID] with the ID of the Cloud Bigtable table to export.
• Replace [APPLICATION_PROFILE_ID] with the ID of the Cloud Bigtable application profile to be used for the export.
• Replace [DESTINATION_PATH] with Cloud Storage path where data should be written. For example, gs://mybucket/somefolder.
• Replace [FILENAME_PREFIX] with the prefix of the SequenceFile file name. For example, output-.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/ \
    --parameters bigtableProject=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],bigtableAppProfileId=[APPLICATION_PROFILE_ID],destinationPath=[DESTINATION_PATH],filenamePrefix=[FILENAME_PREFIX]

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

To run this template with a REST API request, send an HTTP POST request with your project ID. This request requires authorization.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace [PROJECT_ID] with the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
• Replace [INSTANCE_ID] with the ID of the Cloud Bigtable instance that contains the table.
• Replace [TABLE_ID] with the ID of the Cloud Bigtable table to export.
• Replace [APPLICATION_PROFILE_ID] with the ID of the Cloud Bigtable application profile to be used for the export.
• Replace [DESTINATION_PATH] with Cloud Storage path where data should be written. For example, gs://mybucket/somefolder.
• Replace [FILENAME_PREFIX] with the prefix of the SequenceFile file name. For example, output-.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProject": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "bigtableAppProfileId": "[APPLICATION_PROFILE_ID]",
       "destinationPath": "[DESTINATION_PATH]",
       "filenamePrefix": "[FILENAME_PREFIX]",
   },
   "environment": { "zone": "us-central1-f" }
}

Template source code

Java: SDK 2.x

This template's source code is in the GoogleCloudPlatform/cloud-bigtable-client repository on GitHub.

Datastore to Cloud Storage Text

Beta

This feature is in a pre-release state and might change or have limited support. For more information, see the product launch stages.

The Datastore to Cloud Storage Text template is a batch pipeline that reads Datastore entities and writes them to Cloud Storage as text files. You can provide a function to process each entity as a JSON string. If you don't provide such a function, every line in the output file will be a JSON-serialized entity.

Requirements for this pipeline:

Datastore must be set up in the project prior to running the pipeline.

Template parameters

Parameter	Description
datastoreReadGqlQuery	A GQL query that specifies which entities to grab. For example, SELECT * FROM MyKind.
datastoreReadProjectId	The Google Cloud project ID of the Datastore instance that you want to read data from.
datastoreReadNamespace	The namespace of the requested entities. To use the default namespace, leave this parameter blank.
javascriptTextTransformGcsPath	A Cloud Storage path that contains all your JavaScript code. For example, gs://mybucket/mytransforms/*.js. If you don't want to provide a function, leave this parameter blank.
javascriptTextTransformFunctionName	Name of the JavaScript function to be called. For example, if your JavaScript function is function myTransform(inJson) { ...dostuff...} then the function name is myTransform. If you don't want to provide a function, leave this parameter blank.
textWritePrefix	The Cloud Storage path prefix to specify where the data should be written. For example, gs://mybucket/somefolder/.

Running the Datastore to Cloud Storage Text template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select the Datastore to Cloud Storage Text template from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field. Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/Datastore_to_GCS_Text

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace YOUR_BUCKET_NAME with the name of your Cloud Storage bucket.
• Replace YOUR_DATASTORE_KIND with your the type of your Datastore entities.
• Replace YOUR_DATASTORE_NAMESPACE with the namespace of your Datastore entities.
• Replace YOUR_JAVASCRIPT_FUNCTION with your JavaScript function name.
• Replace PATH_TO_JAVASCRIPT_UDF_FILE with the Cloud Storage path to the .js file containing your JavaScript code.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Datastore_to_GCS_Text \
    --parameters \
datastoreReadGqlQuery="SELECT * FROM YOUR_DATASTORE_KIND",\
datastoreReadProjectId=YOUR_PROJECT_ID,\
datastoreReadNamespace=YOUR_DATASTORE_NAMESPACE,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION,\
textWritePrefix=gs://YOUR_BUCKET_NAME/output/

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/Datastore_to_GCS_Text

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

To run this template with a REST API request, send an HTTP POST request with your project ID. This request requires authorization.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace YOUR_BUCKET_NAME with the name of your Cloud Storage bucket.
• Replace YOUR_DATASTORE_KIND with your the type of your Datastore entities.
• Replace YOUR_DATASTORE_NAMESPACE with the namespace of your Datastore entities.
• Replace YOUR_JAVASCRIPT_FUNCTION with your JavaScript function name.
• Replace PATH_TO_JAVASCRIPT_UDF_FILE with the Cloud Storage path to the .js file containing your JavaScript code.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Datastore_to_GCS_Text
{
   "jobName": "JOB_NAME",
   "parameters": {
       "datastoreReadGqlQuery": "SELECT * FROM YOUR_DATASTORE_KIND"
       "datastoreReadProjectId": "YOUR_PROJECT_ID",
       "datastoreReadNamespace": "YOUR_DATASTORE_NAMESPACE",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION",
       "textWritePrefix": "gs://YOUR_BUCKET_NAME/output/"
   },
   "environment": { "zone": "us-central1-f" }
}

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.templates;

import com.google.cloud.teleport.templates.common.DatastoreConverters.DatastoreReadOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.ReadJsonEntities;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.templates.common.TextConverters.FilesystemWriteOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;

/**
 * Dataflow template which copies Datastore Entities to a Text sink.
 * Text is encoded using JSON encoded entity in the v1/Entity rest format:
 * https://cloud.google.com/datastore/docs/reference/rest/v1/Entity
 */
public class DatastoreToText {

  /**
   * Custom PipelineOptions.
   */
  public interface DatastoreToTextOptions extends
      PipelineOptions,
      DatastoreReadOptions,
      JavascriptTextTransformerOptions,
      FilesystemWriteOptions {}

  /**
   * Runs a pipeline which reads in Entities from Datastore, passes in the JSON encoded Entities
   * to a Javascript UDF, and writes the JSON to TextIO sink.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    DatastoreToTextOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(DatastoreToTextOptions.class);

    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply(ReadJsonEntities.newBuilder()
            .setGqlQuery(options.getDatastoreReadGqlQuery())
            .setProjectId(options.getDatastoreReadProjectId())
            .setNamespace(options.getDatastoreReadNamespace())
            .build())
        .apply(TransformTextViaJavascript.newBuilder()
            .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
            .setFunctionName(options.getJavascriptTextTransformFunctionName())
            .build())
        .apply(TextIO.write()
            .to(options.getTextWritePrefix())
            .withSuffix(".json"));

    pipeline.run();
  }
}

Cloud Spanner to Cloud Storage Avro

The Cloud Spanner to Cloud Storage template is a batch pipeline that exports a whole Cloud Spanner database to Cloud Storage in Avro format. Exporting a Cloud Spanner database creates a folder in the bucket you select. The folder contains:

• A spanner-export.json file.
• A TableName-manifest.json file for each table in the database you exported.
• One or more TableName.avro-#####-of-##### files.

For example, exporting a database with two tables, Singers and Albums, creates the following file set:

• Albums-manifest.json
• Albums.avro-00000-of-00002
• Albums.avro-00001-of-00002
• Singers-manifest.json
• Singers.avro-00000-of-00003
• Singers.avro-00001-of-00003
• Singers.avro-00002-of-00003
• spanner-export.json

Requirements for this pipeline:

• The Cloud Spanner database must exist.
• The output Cloud Storage bucket must exist.
• In addition to the Cloud IAM roles necessary to run Dataflow jobs, you must also have the appropriate Cloud IAM roles for reading your Cloud Spanner data and writing to your Cloud Storage bucket.

Template parameters

Parameter	Description
instanceId	The instance ID of the Cloud Spanner database that you want to export.
databaseId	The database ID of the Cloud Spanner database that you want to export.
outputDir	The Cloud Storage path you want to export Avro files to. The export job creates a new directory under this path that contains the exported files.

Running the template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select the Cloud Spanner to Cloud Storage Avro template from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field.
   • Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
   • The job name must match the format cloud-spanner-export-[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] to show up in the Cloud Spanner portion of the Cloud Console.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [DATAFLOW_REGION] with the region where you want the Dataflow job to run (such as us-central1).
• Replace [YOUR_INSTANCE_ID] with your Cloud Spanner instance ID.
• Replace [YOUR_DATABASE_ID] with your Cloud Spanner database ID.
• Replace [YOUR_GCS_DIRECTORY] with the Cloud Storage path that the Avro files should be exported to.
• Replace [JOB_NAME] with a job name of your choice.
  • The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
  • The job name must match the format cloud-spanner-export-[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] to show up in the Cloud Spanner portion of the Cloud Console.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location='gs://dataflow-templates/[VERSION]/Cloud_Spanner_to_GCS_Avro' \
    --region=[DATAFLOW_REGION] \
    --parameters='instanceId=[YOUR_INSTANCE_ID],databaseId=[YOUR_DATABASE_ID],outputDir=[YOUR_GCS_DIRECTORY]

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [DATAFLOW_REGION] with the region where you want the Dataflow job to run (such as us-central1).
• Replace [YOUR_INSTANCE_ID] with your Cloud Spanner instance ID.
• Replace [YOUR_DATABASE_ID] with your Cloud Spanner database ID.
• Replace [YOUR_GCS_DIRECTORY] with the Cloud Storage path that the Avro files should be exported to.
• Replace [JOB_NAME] with a job name of your choice.
  • The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
  • The job name must match the format cloud-spanner-export-[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] to show up in the Cloud Spanner portion of the Cloud Console.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/locations/[DATAFLOW_REGION]/templates:launch?gcsPath=gs://dataflow-templates/[VERSION]/Cloud_Spanner_to_GCS_Avro
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "instanceId": "[YOUR_INSTANCE_ID]",
       "databaseId": "[YOUR_DATABASE_ID]",
       "outputDir": "gs://[YOUR_GCS_DIRECTORY]"
   }
}

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.spanner;

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.gcp.spanner.SpannerConfig;
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.options.ValueProvider;

/** Dataflow template that exports a Cloud Spanner database to Avro files in GCS. */
public class ExportPipeline {

  /** Options for Export pipeline. */
  public interface ExportPipelineOptions extends PipelineOptions {
    @Description("Instance ID for Cloud Spanner")
    ValueProvider<String> getInstanceId();

    void setInstanceId(ValueProvider<String> value);

    @Description("Database ID for Cloud Spanner")
    ValueProvider<String> getDatabaseId();

    void setDatabaseId(ValueProvider<String> value);

    @Description("Output directory in GCS")
    ValueProvider<String> getOutputDir();

    void setOutputDir(ValueProvider<String> value);

    @Description("Test dataflow job identifier for Beam Direct Runner")
    @Default.String(value = "")
    ValueProvider<String> getTestJobId();

    void setTestJobId(ValueProvider<String> jobId);

    @Description("Cloud Spanner API endpoint")
    @Default.String("https://batch-spanner.googleapis.com")
    ValueProvider<String> getSpannerHost();

    void setSpannerHost(ValueProvider<String> value);

    @Description("If true, wait for job finish")
    @Default.Boolean(true)
    boolean getWaitUntilFinish();

    void setWaitUntilFinish(boolean value);
  }

  /**
   * Runs a pipeline to export a Cloud Spanner database to Avro files.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {

    ExportPipelineOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(ExportPipelineOptions.class);

    Pipeline p = Pipeline.create(options);

    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withHost(options.getSpannerHost())
            .withInstanceId(options.getInstanceId())
            .withDatabaseId(options.getDatabaseId());
    p.begin()
        .apply(
            "Run Export",
            new ExportTransform(spannerConfig, options.getOutputDir(), options.getTestJobId()));
    PipelineResult result = p.run();
    if (options.getWaitUntilFinish() &&
        /* Only if template location is null, there is a dataflow job to wait for. Else it's
         * template generation which doesn't start a dataflow job.
         */
        options.as(DataflowPipelineOptions.class).getTemplateLocation() == null) {
      result.waitUntilFinish();
    }
  }
}

Cloud Spanner to Cloud Storage Text

Beta

This feature is in a pre-release state and might change or have limited support. For more information, see the product launch stages.

The Cloud Spanner to Cloud Storage Text template is a batch pipeline that reads in data from a Cloud Spanner table, optionally transforms the data via a JavaScript User Defined Function (UDF) that you provide and writes it to Cloud Storage as CSV text files.

Requirements for this pipeline:

• The input Cloud Spanner table must exist prior to running the pipeline.

Template parameters

Parameter	Description
spannerProjectId	The GCP Project Id of the Cloud Spanner database that you want to read data from.
spannerDatabaseId	Database of requested table.
spannerInstanceId	Instance of requested table.
spannerTable	Table to export.
textWritePrefix	Output Directory where output text files will be written. Please add / at the end. For eg: gs://mybucket/somefolder/.
javascriptTextTransformGcsPath	[Optional] A Cloud Storage path that contains all your JavaScript code. For example, gs://mybucket/mytransforms/*.js. If you don't want to provide a function, leave this parameter blank.
javascriptTextTransformFunctionName	[Optional] Name of the JavaScript function to be called. For example, if your JavaScript function is function myTransform(inJson) { ...dostuff...} then the function name is myTransform. If you don't want to provide a function, leave this parameter blank.

Running the Cloud Spanner to Cloud Storage Text template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select the Cloud Spanner to Cloud Storage Text template from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field. Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace YOUR_DATABASE_ID with the Cloud Spanner database id.
• Replace YOUR_BUCKET_NAME with the name of your Cloud Storage bucket.
• Replace YOUR_INSTANCE_ID with the Cloud Spanner instance id.
• Replace YOUR_TABLE_ID with the Cloud Spanner table id.
• Replace PATH_TO_JAVASCRIPT_UDF_FILE with the Cloud Storage path to the .js file containing your JavaScript code.
• Replace YOUR_JAVASCRIPT_FUNCTION with your JavaScript function name.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Spanner_to_GCS_Text \
    --parameters \
spannerProjectId=YOUR_PROJECT_ID,\
spannerDatabaseId=YOUR_DATABASE_ID,\
spannerInstanceId=YOUR_INSTANCE_ID,\
spannerTable=YOUR_TABLE_ID,\
textWritePrefix=gs://YOUR_BUCKET_NAME/output/,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

To run this template with a REST API request, send an HTTP POST request with your project ID. This request requires authorization.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace YOUR_DATABASE_ID with the Cloud Spanner database id.
• Replace YOUR_BUCKET_NAME with the name of your Cloud Storage bucket.
• Replace YOUR_INSTANCE_ID with the Cloud Spanner instance id.
• Replace YOUR_TABLE_ID with the Cloud Spanner table id.
• Replace PATH_TO_JAVASCRIPT_UDF_FILE with the Cloud Storage path to the .js file containing your JavaScript code.
• Replace YOUR_JAVASCRIPT_FUNCTION with your JavaScript function name.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Spanner_to_GCS_Text
{
   "jobName": "JOB_NAME",
   "parameters": {
       "spannerProjectId": "YOUR_PROJECT_ID",
       "spannerDatabaseId": "YOUR_DATABASE_ID",
       "spannerInstanceId": "YOUR_INSTANCE_ID",
       "spannerTable": "YOUR_TABLE_ID",
       "textWritePrefix": "gs://YOUR_BUCKET_NAME/output/",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION"
   },
   "environment": { "zone": "us-central1-f" }
}

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.templates;

import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.templates.common.SpannerConverters;
import com.google.cloud.teleport.templates.common.SpannerConverters.SpannerReadOptions;
import com.google.cloud.teleport.templates.common.TextConverters.FilesystemWriteOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.spanner.ReadOperation;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Dataflow template which copies a Spanner table to a Text sink. It exports a Spanner table using
 * <a href="https://cloud.google.com/spanner/docs/reads#read_data_in_parallel">Batch API</a>,
 * which creates multiple workers in parallel for better performance. The result is written
 * to a CSV file in Google Cloud Storage. The table schema file is saved in json format along with
 * the exported table.
 *
 * <p>Schema file sample: { "id":"INT64", "name":"STRING(MAX)" }
 *
 * <p>A sample run:
 *
 * <pre>
 * mvn compile exec:java \
 *   -Dexec.mainClass=com.google.cloud.teleport.templates.SpannerToText \
 *   -Dexec.args="--runner=DataflowRunner \
 *                --spannerProjectId=projectId \
 *                --gcpTempLocation=gs://gsTmpLocation \
 *                --spannerInstanceId=instanceId \
 *                --spannerDatabaseId=databaseId \
 *                --spannerTable=table_name \
 *                --textWritePrefix=gcsOutputPath"
 * </pre>
 */
public class SpannerToText {

  private static final Logger LOG = LoggerFactory.getLogger(SpannerToText.class);

  /**
   * Custom PipelineOptions.
   */
  public interface SpannerToTextOptions
      extends PipelineOptions,
          SpannerReadOptions,
          JavascriptTextTransformerOptions,
          FilesystemWriteOptions {}

  /**
   * Runs a pipeline which reads in Records from Spanner, passes in the CSV records to a Javascript
   * UDF, and writes the CSV to TextIO sink.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    LOG.info("Starting pipeline setup");
    PipelineOptionsFactory.register(SpannerToTextOptions.class);
    SpannerToTextOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(SpannerToTextOptions.class);

    FileSystems.setDefaultPipelineOptions(options);
    Pipeline pipeline = Pipeline.create(options);

    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withProjectId(options.getSpannerProjectId())
            .withInstanceId(options.getSpannerInstanceId())
            .withDatabaseId(options.getSpannerDatabaseId());

    PTransform<PBegin, PCollection<ReadOperation>> spannerExport =
        SpannerConverters.ExportTransformFactory.create(
            options.getSpannerTable(), spannerConfig, options.getTextWritePrefix());

    PCollection<String> csv =
        pipeline
            .apply("Create export", spannerExport)
            // We need to use SpannerIO.readAll() instead of SpannerIO.read()
            // because ValueProvider parameters such as table name required for SpannerIO.read()
            // can be read only inside DoFn but SpannerIO.read() is of type
            // PTransform<PBegin, Struct>, which prevents prepending it with DoFn that reads these
            // parameters at the pipeline execution time.
            .apply("Read all records", SpannerIO.readAll().withSpannerConfig(spannerConfig))
            .apply(
                "Struct To Csv",
                MapElements.into(TypeDescriptors.strings())
                    .via(struct -> (new SpannerConverters.StructCsvPrinter()).print(struct)));

    if (options.getJavascriptTextTransformGcsPath().isAccessible()) {
      // The UDF function takes a CSV row as an input and produces a transformed CSV row
      csv =
          csv.apply(
              "JavascriptUDF",
              TransformTextViaJavascript.newBuilder()
                  .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                  .setFunctionName(options.getJavascriptTextTransformFunctionName())
                  .build());
    }
    csv.apply(
        "Write to storage", TextIO.write().to(options.getTextWritePrefix()).withSuffix(".csv"));

    pipeline.run();
    LOG.info("Completed pipeline setup");
  }
}

Cloud Storage Avro to Cloud Bigtable

The Cloud Storage Avro to Cloud Bigtable template is a pipeline that reads data from Avro files in a Cloud Storage bucket and writes the data to a Cloud Bigtable table. You can use the template to copy data from Cloud Storage to Cloud Bigtable.

Requirements for this pipeline:

• The Cloud Bigtable table must exist and have the same column families as exported in the Avro files.
• The input Avro files must exist in a Cloud Storage bucket prior to running the pipeline.
• Cloud Bigtable expects a specific schema from the input Avro files.

Template parameters

Parameter	Description
bigtableProjectId	The ID of the Google Cloud project of the Cloud Bigtable instance that you want to write data to.
bigtableInstanceId	The ID of the Cloud Bigtable instance that contains the table.
bigtableTableId	The ID of the Cloud Bigtable table to import.
inputFilePattern	Cloud Storage path pattern where data is located. For example, gs://mybucket/somefolder/prefix*.

Running the Cloud Storage Avro file to Cloud Bigtable template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select the Cloud Spanner to Cloud Storage Text template from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field. Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace [PROJECT_ID] with the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
• Replace [INSTANCE_ID] with the ID of the Cloud Bigtable instance that contains the table.
• Replace [TABLE_ID] with the ID of the Cloud Bigtable table to export.
• Replace [INPUT_FILE_PATTERN] with Cloud Storage path pattern where data is located. For example, gs://mybucket/somefolder/prefix*.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/Spanner_to_GCS_Text \
    --parameters bigtableProjectId=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],inputFilePattern=[INPUT_FILE_PATTERN]

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

To run this template with a REST API request, send an HTTP POST request with your project ID. This request requires authorization.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace [PROJECT_ID] with the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
• Replace [INSTANCE_ID] with the ID of the Cloud Bigtable instance that contains the table.
• Replace [TABLE_ID] with the ID of the Cloud Bigtable table to export.
• Replace [INPUT_FILE_PATTERN] with Cloud Storage path pattern where data is located. For example, gs://mybucket/somefolder/prefix*.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/Spanner_to_GCS_Text
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProjectId": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "inputFilePattern": "[INPUT_FILE_PATTERN]",
   },
   "environment": { "zone": "us-central1-f" }
}

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.bigtable;

import com.google.bigtable.v2.Mutation;
import com.google.bigtable.v2.Mutation.SetCell;
import com.google.common.collect.ImmutableList;
import com.google.protobuf.ByteString;
import java.nio.ByteBuffer;
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.AvroIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
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.options.ValueProvider;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.KV;

/**
 * Dataflow pipeline that imports data from Avro files in GCS to a Cloud Bigtable table. The Cloud
 * Bigtable table must be created before running the pipeline and must have a compatible table
 * schema. For example, if {@link BigtableCell} from the Avro files has a 'family' of "f1", the
 * Bigtable table should have a column family of "f1".
 */
final class AvroToBigtable {

  /** Options for the import pipeline. */
  public interface Options extends PipelineOptions {
    @Description("The project that contains the table to import into.")
    ValueProvider<String> getBigtableProjectId();

    @SuppressWarnings("unused")
    void setBigtableProjectId(ValueProvider<String> projectId);

    @Description("The Bigtable instance id that contains the table to import into.")
    ValueProvider<String> getBigtableInstanceId();

    @SuppressWarnings("unused")
    void setBigtableInstanceId(ValueProvider<String> instanceId);

    @Description("The Bigtable table id to import into.")
    ValueProvider<String> getBigtableTableId();

    @SuppressWarnings("unused")
    void setBigtableTableId(ValueProvider<String> tableId);

    @Description(
        "The input file patterm to read from. (e.g. gs://mybucket/somefolder/table1*.avro)")
    ValueProvider<String> getInputFilePattern();

    @SuppressWarnings("unused")
    void setInputFilePattern(ValueProvider<String> inputFilePattern);
  }

  /**
   * Runs a pipeline to import Avro files in GCS to a Cloud Bigtable table.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    PipelineResult result = run(options);

    // Wait for pipeline to finish only if it is not constructing a template.
    if (options.as(DataflowPipelineOptions.class).getTemplateLocation() == null) {
      result.waitUntilFinish();
    }
  }

  public static PipelineResult run(Options options) {
    Pipeline pipeline = Pipeline.create(options);

    BigtableIO.Write write =
        BigtableIO.write()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId());

    pipeline
        .apply("Read from Avro", AvroIO.read(BigtableRow.class).from(options.getInputFilePattern()))
        .apply("Transform to Bigtable", MapElements.via(new AvroToBigtableFn()))
        .apply("Write to Bigtable", write);

    return pipeline.run();
  }

  /**
   * Translates {@link BigtableRow} to {@link Mutation}s along with a row key. The mutations are
   * {@link SetCell}s that set the value for specified cells with family name, column qualifier and
   * timestamp.
   */
  static class AvroToBigtableFn
      extends SimpleFunction<BigtableRow, KV<ByteString, Iterable<Mutation>>> {
    @Override
    public KV<ByteString, Iterable<Mutation>> apply(BigtableRow row) {
      ByteString key = toByteString(row.getKey());
      // BulkMutation doesn't split rows. Currently, if a single row contains more than 100,000
      // mutations, the service will fail the request.
      ImmutableList.Builder<Mutation> mutations = ImmutableList.builder();
      for (BigtableCell cell : row.getCells()) {
        SetCell setCell =
            SetCell.newBuilder()
                .setFamilyName(cell.getFamily().toString())
                .setColumnQualifier(toByteString(cell.getQualifier()))
                .setTimestampMicros(cell.getTimestamp())
                .setValue(toByteString(cell.getValue()))
                .build();
        mutations.add(Mutation.newBuilder().setSetCell(setCell).build());
      }
      return KV.of(key, mutations.build());
    }
  }

  /** Copies the content in {@code byteBuffer} into a {@link ByteString}. */
  private static ByteString toByteString(ByteBuffer byteBuffer) {
    return ByteString.copyFrom(byteBuffer.array());
  }
}

Cloud Storage Avro to Cloud Spanner

The Cloud Storage Avro files to Cloud Spanner template is a batch pipeline that reads Avro files from Cloud Storage and imports them to a Cloud Spanner database.

Requirements for this pipeline:

• The target Cloud Spanner database must exist and must be empty.
• You must have read permissions for the Cloud Storage bucket and write permissions for the target Cloud Spanner database.
• The input Cloud Storage path must exist, and it must include a spanner-export.json file that contains a JSON description of files to import.

Warning: If the target Cloud Spanner database is not empty, new data might overwrite pre-existing data, or the import might fail.

Template parameters

Parameter	Description
instanceId	The instance ID of the Cloud Spanner database.
databaseId	The database ID of the Cloud Spanner database.
inputDir	The Cloud Storage path where the Avro files should be imported from.

Running the template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select the Cloud Storage Avro to Cloud Spanner template from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field.
   • Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
   • The job name must match the format cloud-spanner-import-[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] to show up in the Cloud Spanner portion of the Cloud Console.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace [PROJECT_ID] with the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
• Replace [INSTANCE_ID] with the ID of the Cloud Bigtable instance that contains the table.
• Replace [TABLE_ID] with the ID of the Cloud Bigtable table to export.
• Replace [INPUT_FILE_PATTERN] with Cloud Storage path pattern where data is located. For example, gs://mybucket/somefolder/prefix*.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location='gs://dataflow-templates/[VERSION]/GCS_Avro_to_Cloud_Spanner' \
    --region=[DATAFLOW_REGION] \
    --parameters='instanceId=[YOUR_INSTANCE_ID],databaseId=[YOUR_DATABASE_ID],inputDir=[YOUR_GCS_DIRECTORY]'

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace [PROJECT_ID] with the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
• Replace [INSTANCE_ID] with the ID of the Cloud Bigtable instance that contains the table.
• Replace [TABLE_ID] with the ID of the Cloud Bigtable table to export.
• Replace [INPUT_FILE_PATTERN] with Cloud Storage path pattern where data is located. For example, gs://mybucket/somefolder/prefix*.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/locations/[DATAFLOW_REGION]/templates:launch?gcsPath=gs://dataflow-templates/[VERSION]/GCS_Avro_to_Cloud_Spanner
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "instanceId": "[YOUR_INSTANCE_ID]",
       "databaseId": "[YOUR_DATABASE_ID]",
       "inputDir": "gs://[YOUR_GCS_DIRECTORY]"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.spanner;

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.gcp.spanner.SpannerConfig;
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.options.ValueProvider;

/** Avro to Cloud Spanner Import pipeline. */
public class ImportPipeline {

  /** Options for {@link ImportPipeline}. */
  public interface Options extends PipelineOptions {

    @Description("Instance ID to write to Spanner")
    ValueProvider<String> getInstanceId();

    void setInstanceId(ValueProvider<String> value);

    @Description("Database ID to write to Spanner")
    ValueProvider<String> getDatabaseId();

    void setDatabaseId(ValueProvider<String> value);

    @Description("Input directory")
    ValueProvider<String> getInputDir();

    void setInputDir(ValueProvider<String> value);

    @Description("Spanner host")
    @Default.String("https://batch-spanner.googleapis.com")
    ValueProvider<String> getSpannerHost();

    void setSpannerHost(ValueProvider<String> value);

    @Description("If true wait until indexes are finished. Useful for testing.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getWaitForIndexes();

    void setWaitForIndexes(ValueProvider<Boolean> value);

    @Description("If true, wait for job finish")
    @Default.Boolean(true)
    boolean getWaitUntilFinish();

    void setWaitUntilFinish(boolean value);
  }

  public static void main(String[] args) {

    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    Pipeline p = Pipeline.create(options);

    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withHost(options.getSpannerHost())
            .withInstanceId(options.getInstanceId())
            .withDatabaseId(options.getDatabaseId());

    p.apply(new ImportTransform(spannerConfig, options.getInputDir(), options.getWaitForIndexes()));

    PipelineResult result = p.run();
    if (options.getWaitUntilFinish() &&
        /* Only if template location is null, there is a dataflow job to wait for. Else it's
         * template generation which doesn't start a dataflow job.
         */
        options.as(DataflowPipelineOptions.class).getTemplateLocation() == null) {
      result.waitUntilFinish();
    }
  }
}

Cloud Storage SequenceFile to Cloud Bigtable

The Cloud Storage SequenceFile to Cloud Bigtable template is a pipeline that reads data from SequenceFiles in a Cloud Storage bucket and writes the data to a Cloud Bigtable table. You can use the template to copy data from Cloud Storage to Cloud Bigtable.

Requirements for this pipeline:

• The Cloud Bigtable table must exist.
• The input SequenceFiles must exist in a Cloud Storage bucket prior to running the pipeline.
• The input SequenceFiles must have been exported from Cloud Bigtable or HBase.

Template parameters

Parameter	Description
bigtableProject	The ID of the Google Cloud project of the Cloud Bigtable instance that you want to write data to.
bigtableInstanceId	The ID of the Cloud Bigtable instance that contains the table.
bigtableTableId	The ID of the Cloud Bigtable table to import.
bigtableAppProfileId	The ID of the Cloud Bigtable application profile to be used for the import. If you do not specify an app profile, Cloud Bigtable uses the instance's default app profile.
sourcePattern	Cloud Storage path pattern where data is located. For example, gs://mybucket/somefolder/prefix*.

Running the Cloud Storage SequenceFile to Cloud Bigtable template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select the Cloud Storage Avro to Cloud Spanner template from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field. Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace [PROJECT_ID] with the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
• Replace [INSTANCE_ID] with the ID of the Cloud Bigtable instance that contains the table.
• Replace [TABLE_ID] with the ID of the Cloud Bigtable table to export.
• Replace [APPLICATION_PROFILE_ID] with the ID of the Cloud Bigtable application profile to be used for the export.
• Replace [SOURCE_PATTERN] with Cloud Storage path pattern where data is located. For example, gs://mybucket/somefolder/prefix*.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/GCS_Avro_to_Cloud_Spanner \
    --parameters bigtableProject=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],bigtableAppProfileId=[APPLICATION_PROFILE_ID],sourcePattern=[SOURCE_PATTERN]

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

To run this template with a REST API request, send an HTTP POST request with your project ID. This request requires authorization.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace [PROJECT_ID] with the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from.
• Replace [INSTANCE_ID] with the ID of the Cloud Bigtable instance that contains the table.
• Replace [TABLE_ID] with the ID of the Cloud Bigtable table to export.
• Replace [APPLICATION_PROFILE_ID] with the ID of the Cloud Bigtable application profile to be used for the export.
• Replace [SOURCE_PATTERN] with Cloud Storage path pattern where data is located. For example, gs://mybucket/somefolder/prefix*.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_Avro_to_Cloud_Spanner
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProject": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "bigtableAppProfileId": "[APPLICATION_PROFILE_ID]",
       "sourcePattern": "[SOURCE_PATTERN]",
   },
   "environment": { "zone": "us-central1-f" }
}

Template source code

Java: SDK 2.x

This template's source code is in the GoogleCloudPlatform/cloud-bigtable-client repository on GitHub.

Cloud Storage Text to BigQuery

Beta

This feature is in a pre-release state and might change or have limited support. For more information, see the product launch stages.

The Cloud Storage Text to BigQuery pipeline is a batch pipeline that allows you to read text files stored in Cloud Storage, transform them using a JavaScript User Defined Function (UDF) that you provide, and output the result to BigQuery.

IMPORTANT: If you reuse an existing BigQuery table, the table will be overwritten.

Requirements for this pipeline:

• Create a JSON file that describes your BigQuery schema.

  Ensure that there is a top level JSON array titled BigQuery Schema and that its contents follow the pattern {"name": "COLUMN_NAME", "type": "DATA_TYPE"}. For example:

  {
    "BigQuery Schema": [
      {
        "name": "location",
        "type": "STRING"
      },
      {
        "name": "name",
        "type": "STRING"
      },
      {
        "name": "age",
        "type": "STRING"
      },
      {
        "name": "color",
        "type": "STRING"
      },
      {
        "name": "coffee",
        "type": "STRING"
      }
    ]
  }
  

• Create a JavaScript (.js) file with your UDF function that supplies the logic to transform the lines of text. Note that your function must return a JSON string.

  For example, this function splits each line of a CSV file and returns a JSON string after transforming the values.

  function transform(line) {
  var values = line.split(',');
  
  var obj = new Object();
  obj.location = values[0];
  obj.name = values[1];
  obj.age = values[2];
  obj.color = values[3];
  obj.coffee = values[4];
  var jsonString = JSON.stringify(obj);
  
  return jsonString;
  }
  

Template parameters

Parameter	Description
javascriptTextTransformFunctionName	The name of the function you want to call from your .js file.
JSONPath	The gs:// path to the JSON file that defines your BigQuery schema, stored in Cloud Storage. For example, gs://path/to/my/schema.json.
javascriptTextTransformGcsPath	The gs:// path to the JavaScript file that defines your UDF. For example, gs://path/to/my/javascript_function.js.
inputFilePattern	The gs:// path to the text in Cloud Storage you'd like to process. For example, gs://path/to/my/text/data.txt.
outputTable	The BigQuery table name you want to create to store your processed data in. If you reuse an existing BigQuery table, the table will be overwritten. For example, my-project-name:my-dataset.my-table.
bigQueryLoadingTemporaryDirectory	Temporary directory for BigQuery loading process. For example, gs://my-bucket/my-files/temp_dir.

Running the Cloud Storage Text to BigQuery template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select the Cloud Storage Text to BigQuery template from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field. Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Text_to_BigQuery

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace YOUR_JAVASCRIPT_FUNCTION with the name of your UDF.
• Replace PATH_TO_BIGQUERY_SCHEMA_JSON with the Cloud Storage path to the JSON file containing the schema definition.
• Replace PATH_TO_JAVASCRIPT_UDF_FILE with the Cloud Storage path to the .js file containing your JavaScript code.
• Replace PATH_TO_YOUR_TEXT_DATA with your Cloud Storage path to your text dataset.
• Replace BIGQUERY_TABLE with your BigQuery table name.
• Replace PATH_TO_TEMP_DIR_ON_GCS with your Cloud Storage path to the temp directory.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/GCS_Text_to_BigQuery \
    --parameters \
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION,\
JSONPath=PATH_TO_BIGQUERY_SCHEMA_JSON,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
inputFilePattern=PATH_TO_YOUR_TEXT_DATA,\
outputTable=BIGQUERY_TABLE,\
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Text_to_BigQuery

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

To run this template with a REST API request, send an HTTP POST request with your project ID. This request requires authorization.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace YOUR_JAVASCRIPT_FUNCTION with the name of your UDF.
• Replace PATH_TO_BIGQUERY_SCHEMA_JSON with the Cloud Storage path to the JSON file containing the schema definition.
• Replace PATH_TO_JAVASCRIPT_UDF_FILE with the Cloud Storage path to the .js file containing your JavaScript code.
• Replace PATH_TO_YOUR_TEXT_DATA with your Cloud Storage path to your text dataset.
• Replace BIGQUERY_TABLE with your BigQuery table name.
• Replace PATH_TO_TEMP_DIR_ON_GCS with your Cloud Storage path to the temp directory.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_Text_to_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION",
       "JSONPath": "PATH_TO_BIGQUERY_SCHEMA_JSON",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "inputFilePattern":"PATH_TO_YOUR_TEXT_DATA",
       "outputTable":"BIGQUERY_TABLE",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS"
   },
   "environment": { "zone": "us-central1-f" }
}

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */


package com.google.cloud.teleport.templates;

import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableSchema;
import com.google.cloud.teleport.templates.common.BigQueryConverters;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.json.JSONArray;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Templated pipeline to read text from TextIO, apply a javascript UDF to it, and write it to GCS.
 */
public class TextIOToBigQuery {

  /** Options supported by {@link TextIOToBigQuery}. */
  public interface Options extends DataflowPipelineOptions, JavascriptTextTransformerOptions {
    @Description("The GCS location of the text you'd like to process")
    ValueProvider<String> getInputFilePattern();

    void setInputFilePattern(ValueProvider<String> value);

    @Description("JSON file with BigQuery Schema description")
    ValueProvider<String> getJSONPath();

    void setJSONPath(ValueProvider<String> value);

    @Description("Output topic to write to")
    ValueProvider<String> getOutputTable();

    void setOutputTable(ValueProvider<String> value);

    @Description("GCS path to javascript fn for transforming output")
    ValueProvider<String> getJavascriptTextTransformGcsPath();

    void setJavascriptTextTransformGcsPath(ValueProvider<String> jsTransformPath);

    @Validation.Required
    @Description("UDF Javascript Function Name")
    ValueProvider<String> getJavascriptTextTransformFunctionName();

    void setJavascriptTextTransformFunctionName(
        ValueProvider<String> javascriptTextTransformFunctionName);

    @Validation.Required
    @Description("Temporary directory for BigQuery loading process")
    ValueProvider<String> getBigQueryLoadingTemporaryDirectory();

    void setBigQueryLoadingTemporaryDirectory(ValueProvider<String> directory);
  }

  private static final Logger LOG = LoggerFactory.getLogger(TextIOToBigQuery.class);

  private static final String BIGQUERY_SCHEMA = "BigQuery Schema";
  private static final String NAME = "name";
  private static final String TYPE = "type";
  private static final String MODE = "mode";

  public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply("Read from source", TextIO.read().from(options.getInputFilePattern()))
        .apply(
            TransformTextViaJavascript.newBuilder()
                .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                .setFunctionName(options.getJavascriptTextTransformFunctionName())
                .build())
        .apply(BigQueryConverters.jsonToTableRow())
        .apply(
            "Insert into Bigquery",
            BigQueryIO.writeTableRows()
                .withSchema(
                    NestedValueProvider.of(
                        options.getJSONPath(),
                        new SerializableFunction<String, TableSchema>() {

                          @Override
                          public TableSchema apply(String jsonPath) {

                            TableSchema tableSchema = new TableSchema();
                            List<TableFieldSchema> fields = new ArrayList<>();
                            SchemaParser schemaParser = new SchemaParser();
                            JSONObject jsonSchema;

                            try {

                              jsonSchema = schemaParser.parseSchema(jsonPath);

                              JSONArray bqSchemaJsonArray =
                                  jsonSchema.getJSONArray(BIGQUERY_SCHEMA);

                              for (int i = 0; i < bqSchemaJsonArray.length(); i++) {
                                JSONObject inputField = bqSchemaJsonArray.getJSONObject(i);
                                TableFieldSchema field =
                                    new TableFieldSchema()
                                        .setName(inputField.getString(NAME))
                                        .setType(inputField.getString(TYPE));

                                if (inputField.has(MODE)) {
                                  field.setMode(inputField.getString(MODE));
                                }

                                fields.add(field);
                              }
                              tableSchema.setFields(fields);

                            } catch (Exception e) {
                              throw new RuntimeException(e);
                            }
                            return tableSchema;
                          }
                        }))
                .to(options.getOutputTable())
                .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
                .withWriteDisposition(WriteDisposition.WRITE_TRUNCATE)
                .withCustomGcsTempLocation(options.getBigQueryLoadingTemporaryDirectory()));

    pipeline.run();
  }
}

Cloud Storage Text to Datastore

Beta

This feature is in a pre-release state and might change or have limited support. For more information, see the product launch stages.

The Cloud Storage Text to Datastore template is a batch pipeline which reads from text files stored in Cloud Storage and writes JSON encoded Entities to Datastore. Each line in the input text files should be in JSON format specified in https://cloud.google.com/datastore/docs/reference/rest/v1/Entity .

Requirements for this pipeline:

• Datastore must be enabled in the destination project.

Template parameters

Parameter	Description
textReadPattern	A Cloud Storage file path pattern that specifies the location of your text data files. For example, gs://mybucket/somepath/*.json.
javascriptTextTransformGcsPath	A Cloud Storage path pattern that contains all your JavaScript code. For example, gs://mybucket/mytransforms/*.js. If you don't want to provide a function, leave this parameter blank.
javascriptTextTransformFunctionName	Name of the JavaScript function to be called. For example, if your JavaScript function is function myTransform(inJson) { ...dostuff...} then the function name is myTransform. If you don't want to provide a function, leave this parameter blank.
datastoreWriteProjectId	The Google Cloud project id of where to write the Datastore entities
errorWritePath	The error log output file to use for write failures that occur during processing. For example, gs://bucket-name/errors.txt.

Running the Cloud Storage Text to Datastore template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select the Cloud Storage Text to Datastore template from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field. Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Text_to_Datastore

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace PATH_TO_INPUT_TEXT_FILES with the input files pattern on Cloud Storage.
• Replace YOUR_JAVASCRIPT_FUNCTION with your JavaScript function name.
• Replace PATH_TO_JAVASCRIPT_UDF_FILE with the Cloud Storage path to the .js file containing your JavaScript code.
• Replace ERROR_FILE_WRITE_PATH with your desired path to error file on Cloud Storage.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/GCS_Text_to_Datastore \
    --parameters \
textReadPattern=PATH_TO_INPUT_TEXT_FILES,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION,\
datastoreWriteProjectId=YOUR_PROJECT_ID,\
errorWritePath=ERROR_FILE_WRITE_PATH

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Text_to_Datastore

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

To run this template with a REST API request, send an HTTP POST request with your project ID. This request requires authorization.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace PATH_TO_INPUT_TEXT_FILES with the input files pattern on Cloud Storage.
• Replace YOUR_JAVASCRIPT_FUNCTION with your JavaScript function name.
• Replace PATH_TO_JAVASCRIPT_UDF_FILE with the Cloud Storage path to the .js file containing your JavaScript code.
• Replace ERROR_FILE_WRITE_PATH with your desired path to error file on Cloud Storage.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_Text_to_Datastore
{
   "jobName": "JOB_NAME",
   "parameters": {
       "textReadPattern": "PATH_TO_INPUT_TEXT_FILES",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION",
       "datastoreWriteProjectId": "YOUR_PROJECT_ID",
       "errorWritePath": "ERROR_FILE_WRITE_PATH"
   },
   "environment": { "zone": "us-central1-f" }
}

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.templates;

import com.google.cloud.teleport.templates.common.DatastoreConverters.DatastoreWriteOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.WriteJsonEntities;
import com.google.cloud.teleport.templates.common.ErrorConverters.ErrorWriteOptions;
import com.google.cloud.teleport.templates.common.ErrorConverters.LogErrors;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.templates.common.TextConverters.FilesystemReadOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.values.TupleTag;

/**
 * Dataflow template which reads from a Text Source and writes JSON encoded Entities into Datastore.
 * The Json is expected to be in the format of:
 * https://cloud.google.com/datastore/docs/reference/rest/v1/Entity
 */
public class TextToDatastore {

  /**
   * TextToDatastore Pipeline Options.
   */
  public interface TextToDatastoreOptions extends
      PipelineOptions,
      FilesystemReadOptions,
      JavascriptTextTransformerOptions,
      DatastoreWriteOptions,
      ErrorWriteOptions {}

  /**
   * Runs a pipeline which reads from a Text Source, passes the Text to a Javascript UDF,
   * writes the JSON encoded Entities to a TextIO sink.
   *
   * <p>If your Text Source does not contain JSON encoded Entities, then you'll need to supply a
   * Javascript UDF which transforms your data to be JSON encoded Entities.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    TextToDatastoreOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(TextToDatastoreOptions.class);

    TupleTag<String> errorTag = new TupleTag<String>("errors"){};

    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply(TextIO.read()
            .from(options.getTextReadPattern()))
        .apply(TransformTextViaJavascript.newBuilder()
            .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
            .setFunctionName(options.getJavascriptTextTransformFunctionName())
            .build())
        .apply(WriteJsonEntities.newBuilder()
            .setProjectId(options.getDatastoreWriteProjectId())
            .setErrorTag(errorTag)
            .build())
        .apply(LogErrors.newBuilder()
            .setErrorWritePath(options.getErrorWritePath())
            .setErrorTag(errorTag)
            .build());

    pipeline.run();
  }
}

Cloud Storage Text to Pub/Sub (Batch)

Beta

This feature is in a pre-release state and might change or have limited support. For more information, see the product launch stages.

This template creates a batch pipeline that reads records from text files stored in Cloud Storage and publishes them to a Pub/Sub topic. The template can be used to publish records in a newline-delimited file containing JSON records or CSV file to a Pub/Sub topic for real-time processing. You can use this template to replay data to Pub/Sub.

Note that this template does not set any timestamp on the individual records, so the event time will be equal to the publishing time during execution. If your pipeline is reliant on an accurate event time for processing, you should not use this pipeline.

Requirements for this pipeline:

• The files to read need to be in newline-delimited JSON or CSV format. Records spanning multiple lines in the source files may cause issues downstream as each line within the files will be published as a message to Pub/Sub.
• The Pub/Sub topic must exist prior to running the pipeline.

Template parameters

Parameter	Description
inputFilePattern	The input file pattern to read from. For example, gs://bucket-name/files/*.json.
outputTopic	The Pub/Sub input topic to write to. The name should be in the format of projects/<project-id>/topics/<topic-name>.

Running the Cloud Storage Text to Pub/Sub (Batch) template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select the Cloud Storage Text to Pub/Sub (Batch) template from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field. Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_PubSub

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace YOUR_TOPIC_NAME with your Pub/Sub topic name.
• Replace YOUR_BUCKET_NAME with the name of your Cloud Storage bucket.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub \
    --parameters \
inputFilePattern=gs://YOUR_BUCKET_NAME/files/*.json,\
outputTopic=projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_PubSub

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

To run this template with a REST API request, send an HTTP POST request with your project ID. This request requires authorization.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace YOUR_TOPIC_NAME with your Pub/Sub topic name.
• Replace YOUR_BUCKET_NAME with the name of your Cloud Storage bucket.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputFilePattern": "gs://YOUR_BUCKET_NAME/files/*.json",
       "outputTopic": "projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME"
   },
   "environment": { "zone": "us-central1-f" }
}

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */


package com.google.cloud.teleport.templates;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
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.options.Validation.Required;
import org.apache.beam.sdk.options.ValueProvider;

/**
 * The {@code TextToPubsub} pipeline publishes records to
 * Cloud Pub/Sub from a set of files. The pipeline reads each
 * file row-by-row and publishes each record as a string message.
 * At the moment, publishing messages with attributes is unsupported.
 *
 * <p>Example Usage:
 *
 * <pre>
 * {@code mvn compile exec:java \
    -Dexec.mainClass=com.google.cloud.teleport.templates.TextToPubsub \
    -Dexec.args=" \
    --project=${PROJECT_ID} \
    --stagingLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/staging \
    --tempLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/temp \
    --runner=DataflowRunner \
    --inputFilePattern=gs://path/to/demo_file.csv \
    --outputTopic=projects/${PROJECT_ID}/topics/${TOPIC_NAME}"
 * }
 * </pre>
 *
 */
public class TextToPubsub {

  /**
   * The custom options supported by the pipeline. Inherits
   * standard configuration options.
   */
  public interface Options extends PipelineOptions {
    @Description("The file pattern to read records from (e.g. gs://bucket/file-*.csv)")
    @Required
    ValueProvider<String> getInputFilePattern();
    void setInputFilePattern(ValueProvider<String> value);

    @Description("The name of the topic which data should be published to. "
        + "The name should be in the format of projects/<project-id>/topics/<topic-name>.")
    @Required
    ValueProvider<String> getOutputTopic();
    void setOutputTopic(ValueProvider<String> value);
  }

  /**
   * Main entry-point for the pipeline. Reads in the
   * command-line arguments, parses them, and executes
   * the pipeline.
   *
   * @param args  Arguments passed in from the command-line.
   */
  public static void main(String[] args) {

    // Parse the user options passed from the command-line
    Options options = PipelineOptionsFactory
        .fromArgs(args)
        .withValidation()
        .as(Options.class);

    run(options);
  }

  /**
   * Executes the pipeline with the provided execution
   * parameters.
   *
   * @param options The execution parameters.
   */
  public static PipelineResult run(Options options) {
    // Create the pipeline.
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps:
     *  1) Read from the text source.
     *  2) Write each text record to Pub/Sub
     */
    pipeline
        .apply("Read Text Data", TextIO.read().from(options.getInputFilePattern()))
        .apply("Write to PubSub", PubsubIO.writeStrings().to(options.getOutputTopic()));

    return pipeline.run();
  }
}

Cloud Storage Text to Cloud Spanner

The Cloud Storage Text to Cloud Spanner template is a batch pipeline that reads CSV text files from Cloud Storage and imports them to a Cloud Spanner database.

Requirements for this pipeline:

• The target Cloud Spanner database and table must exist.
• You must have read permissions for the Cloud Storage bucket and write permissions for the target Cloud Spanner database.
• The input Cloud Storage path containing the CSV files must exist.
• You must create an import manifest file containing a JSON description of the CSV files, and you must store that manifest in Cloud Storage.
• If the target Cloud Spanner database already has a schema, any columns specified in the manifest file must have the same data types as their corresponding columns in the target database's schema.

• The manifest file, encoded in ASCII or UTF-8, must match the following format:

  Manifest format and example

  The format of the manifest file corresponds to the following message type, shown here in protocol buffer format:

  message ImportManifest {
    // The per-table import manifest.
    message TableManifest {
      // Required. The name of the destination table.
      string table_name = 1;
      // Required. The CSV files to import. This value can be either a filepath or a glob pattern.
      repeated string file_patterns = 2;
      // The schema for a table column.
      message Column {
        // Required for each Column that you specify. The name of the column in the
        // destination table.
        string column_name = 1;
        // Required for each Column that you specify. The type of the column. Supports the
        // following data types: BOOL, INT64, FLOAT64, STRING, DATE, and TIMESTAMP.
        string type_name = 2;
      }
      // Optional. The schema for the table columns.
      repeated Column columns = 3;
    }
    // Required. The TableManifest of the tables to be imported.
    repeated TableManifest tables = 1;
  }

  The following example shows a manifest file for importing tables called Albums and Singers. The Albums table uses the column schema that the job retrieves from the database, and the Singers table uses the schema that the manifest file specifies:

  {
    "tables": [
      {
        "table_name": "Albums",
        "file_patterns": [
          "gs://bucket1/Albums_1.csv",
          "gs://bucket1/Albums_2.csv"
        ]
      },
      {
        "table_name": "Singers",
        "file_patterns": [
          "gs://bucket1/Singers*.csv"
        ],
        "columns": [
          {"column_name": "SingerId", "type_name": "INT64"},
          {"column_name": "FirstName", "type_name": "STRING"},
          {"column_name": "LastName", "type_name": "STRING"}
        ]
      }
    ]
  }

• Text files to be imported must be in CSV format, with ASCII or UTF-8 encoding. We recommend not using byte order mark (BOM) in UTF-8 encoded files.
• Data must match one of the following types:
  • INT64
  • FLOAT64
  • BOOL
  • STRING
  • DATE
  • TIMESTAMP

Note: If you do not specify the target table's column names and data types in the import manifest file, the columns in the CSV files must be in the same order as the columns in the target database. You can see the order of the columns in your table by running the following query:

SELECT * FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME =
      [YOUR_TABLE_NAME] ORDER BY ORDINAL_POSITION

Warning: If the target Cloud Spanner database is not empty, new data might overwrite pre-existing data.

Template parameters

Parameter	Description
instanceId	The instance ID of the Cloud Spanner database.
databaseId	The database ID of the Cloud Spanner database.
importManifest	The path in Cloud Storage to the import manifest file.
columnDelimiter	The column delimiter that the source file uses. The default value is ,.
fieldQualifier	The character that should surround any value in the source file that contains the columnDelimiter. The default value is ".
trailingDelimiter	Specifies whether the lines in the source files have trailing delimiters (that is, if the columnDelimiter character appears at the end of each line, after the last column value). The default value is true.
escape	The escape character the source file uses. By default, this parameter is not set and the template does not use the escape character.
nullString	The string that represents a NULL value. By default, this parameter is not set and the template does not use the null string.
dateFormat	The format used to parse date columns. By default, the pipeline tries to parse the date columns as yyyy-M-d[' 00:00:00'], for example, as 2019-01-31 or 2019-1-1 00:00:00. If your date format is different, specify the format using the java.time.format.DateTimeFormatter patterns.
timestampFormat	The format used to parse timestamp columns. If the timestamp is a long integer, then it is parsed as Unix epoch. Otherwise, it is parsed as a string using the java.time.format.DateTimeFormatter.ISO_INSTANT format. For other cases, specify your own pattern string, for example, using MMM dd yyyy HH:mm:ss.SSSVV for timestamps in the form of "Jan 21 1998 01:02:03.456+08:00".

If you need to use customized date or timestamp formats, make sure they're valid java.time.format.DateTimeFormatter patterns. The following table shows additional examples of customized formats for date and timestamp columns:

Type	Input value	Format	Remark
DATE	2011-3-31		By default, the template can parse this format. You don't need to specify the dateFormat parameter.
DATE	2011-3-31 00:00:00		By default, the template can parse this format. You don't need to specify the format. If you like, you can use yyyy-M-d' 00:00:00'.
DATE	01 Apr, 18	dd MMM, yy
DATE	Wednesday, April 3, 2019 AD	EEEE, LLLL d, yyyy G
TIMESTAMP	2019-01-02T11:22:33Z 2019-01-02T11:22:33.123Z 2019-01-02T11:22:33.12356789Z		The default format ISO_INSTANT can parse this type of timestamp. You don't need to provide the timestampFormat parameter.
TIMESTAMP	1568402363		By default, the template can parse this type of timestamp and treat it as the Unix epoch time.
TIMESTAMP	Tue, 3 Jun 2008 11:05:30 GMT	EEE, d MMM yyyy HH:mm:ss VV
TIMESTAMP	2018/12/31 110530.123PST	yyyy/MM/dd HHmmss.SSSz
TIMESTAMP	2019-01-02T11:22:33Z or 2019-01-02T11:22:33.123Z	yyyy-MM-dd'T'HH:mm:ss[.SSS]VV	If the input column is a mix 2019-01-02T11:22:33Z and 2019-01-02T11:22:33.123Z, the default format can parse this type of timestamp. You don't need to provide your own format parameter. However, you can use yyyy-MM-dd'T'HH:mm:ss[.SSS]VV to handle both cases. Note that you cannot use yyyy-MM-dd'T'HH:mm:ss[.SSS]'Z', because the postfix 'Z' must be parsed as a time-zone ID, not a character literal. Internally, the timestamp column is converted to a java.time.Instant. Therefore, it must be specified in UTC or have time zone information associated with it. Local datetime, such as 2019-01-02 11:22:33, cannot be parsed as a valid java.time.Instant.

Running the template

Console

Run in the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select the Cloud Storage Text to Cloud Spanner template from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field. Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

gcloud

Run with the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_Spanner

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [DATAFLOW_REGION] with the region where you want the Dataflow job to run (such as us-central1).
• Replace [YOUR_INSTANCE_ID] with your Cloud Spanner instance ID.
• Replace [YOUR_DATABASE_ID] with your Cloud Spanner database ID.
• Replace [GCS_PATH_TO_IMPORT_MANIFEST] with the Cloud Storage path to your import manifest file.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location='gs://dataflow-templates/[VERSION]/GCS_Text_to_Cloud_Spanner' \
    --region=[DATAFLOW_REGION] \
    --parameters='instanceId=[YOUR_INSTANCE_ID],databaseId=[YOUR_DATABASE_ID],importManifest=[GCS_PATH_TO_IMPORT_MANIFEST]'

API

Run with the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_Spanner

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

Use this example request as documented in Using the REST API. This request requires authorization, and you must specify a tempLocation where you have write permissions. You must replace the following values in this example:

• Replace [YOUR_PROJECT_ID] with your project ID.
• Replace [DATAFLOW_REGION] with the region where you want the Dataflow job to run (such as us-central1).
• Replace [YOUR_INSTANCE_ID] with your Cloud Spanner instance ID.
• Replace [YOUR_DATABASE_ID] with your Cloud Spanner database ID.
• Replace [GCS_PATH_TO_IMPORT_MANIFEST] with the Cloud Storage path to your import manifest file.
• Replace [JOB_NAME] with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/locations/[DATAFLOW_REGION]/templates:launch?gcsPath=gs://dataflow-templates/[VERSION]/GCS_Text_to_Cloud_Spanner

{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "instanceId": "[YOUR_INSTANCE_ID]",
       "databaseId": "[YOUR_DATABASE_ID]",
       "importManifest": "[GCS_PATH_TO_IMPORT_MANIFEST]"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}

Template source code

Java: SDK 2.x

This template's source code is in the GoogleCloudPlatform/DataflowTemplates repository on GitHub.

Java Database Connectivity (JDBC) to BigQuery

Beta

This feature is in a pre-release state and might change or have limited support. For more information, see the product launch stages.

The JDBC to BigQuery template is a batch pipeline that copies data from a relational database table into an existing BigQuery table. This pipeline uses JDBC to connect to the relational database. You can use this template to copy data from any relational database with available JDBC drivers into BigQuery. For an extra layer of protection, you can also pass in a KMS key along with a Base64-encoded username, password, and connection string parameters encrypted with the KMS key. See the KMS API encryption endpoint for additional details on encrypting your username, password, and connection string parameters.

Requirements for this pipeline:

• The JDBC drivers for the relational database must be available.
• The BigQuery table must exist prior to pipeline execution.
• The BigQuery table must have a compatible schema.
• The relational database must be accessible from the subnet where Dataflow runs.

Template parameters

Parameter	Description
driverJars	Comma separated list of driver jars. For example, gs://<my-bucket>/driver_jar1.jar,gs://<my-bucket>/driver_jar2.jar.
driverClassName	The JDBC driver class name. For example, com.mysql.jdbc.Driver.
connectionURL	The JDBC connection URL string. For example, jdbc:mysql://some-host:3306/sampledb. Can be passed in as a Base64-encoded string encrypted with a KMS key.
query	Query to be run on the source to extract the data. For example, select * from sampledb.sample_table.
outputTable	The BigQuery output table location, in the format of <my-project>:<my-dataset>.<my-table>.
bigQueryLoadingTemporaryDirectory	Temporary directory for BigQuery loading process. For example, gs://<my-bucket>/my-files/temp_dir.
connectionProperties	[Optional] Properties string to use for the JDBC connection. For example, unicode=true&characterEncoding=UTF-8.
username	[Optional] Username to be used for the JDBC connection. Can be passed in as a Base64encoded string encrypted with a KMS key.
password	[Optional] Password to be used for the JDBC connection. Can be passed in as a Base64-encoded string encrypted with a KMS key.
KMSEncryptionKey	[Optional] KMS Encryption Key to decrypt the username, password, and connection string. If KMS key is passed in, the username, password and connection string must all be passed in encrypted.

Running the JDBC to BigQuery template

CONSOLE

Run from the Google Cloud Console

1. Go to the Dataflow page in the Cloud Console.
Go to the Dataflow page
2. Click Create job from template.



3. Select the JDBC to BigQuery template from the Dataflow template drop-down menu.
4. Enter a job name in the Job Name field. Your job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
5. Enter your parameter values in the provided parameter fields.
6. Click Run Job.

GCLOUD

Run from the gcloud command-line tool

Note: To use the gcloud command-line tool to run templates, you must have Cloud SDK version 138.0.0 or higher.

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/Jdbc_to_BigQuery

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace DRIVER_PATHS with the comma separated Cloud Storage path(s) of the JDBC driver(s).
• Replace DRIVER_CLASS_NAME with the drive class name.
• Replace JDBC_CONNECTION_URL with the JDBC connection URL.
• Replace SOURCE_SQL_QUERY with the SQL query to be run on the source database.
• Replace YOUR_DATASET with your BigQuery dataset, and replace YOUR_TABLE_NAME with your BigQuery table name.
• Replace PATH_TO_TEMP_DIR_ON_GCS with your Cloud Storage path to the temp directory.
• Replace CONNECTION_PROPERTIES with the JDBC connection properties if required.
• Replace CONNECTION_USERNAME with the JDBC connection username.
• Replace CONNECTION_PASSWORD with the JDBC connection password.
• Replace KMS_ENCRYPTION_KEY with the KMS Encryption Key.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Jdbc_to_BigQuery \
    --parameters \
driverJars=DRIVER_PATHS,\
driverClassName=DRIVER_CLASS_NAME,\
connectionURL=JDBC_CONNECTION_URL,\
query=SOURCE_SQL_QUERY,\
outputTable=YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME,
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS,\
connectionProperties=CONNECTION_PROPERTIES,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
KMSEncryptionKey=KMS_ENCRYPTION_KEY

API

Run from the REST API

When running this template, you'll need the Cloud Storage path to the template:

gs://dataflow-templates/VERSION/Jdbc_to_BigQuery

Caution: The latest template version is kept in the non-dated parent folder gs://dataflow-templates/latest which may update with breaking changes. Your production environments should use templates kept in the most recent dated parent folder to prevent these breaking changes from affecting your production workflows. All dated template versions can be found nested inside their dated parent folder in the Cloud Storage bucket: gs://dataflow-templates/.

To run this template with a REST API request, send an HTTP POST request with your project ID. This request requires authorization.

You must replace the following values in this example:

• Replace YOUR_PROJECT_ID with your project ID.
• Replace JOB_NAME with a job name of your choice. The job name must match the regular expression [a-z]([-a-z0-9]{0,38}[a-z0-9])? to be valid.
• Replace DRIVER_PATHS with the comma separated Cloud Storage path(s) of the JDBC driver(s).
• Replace DRIVER_CLASS_NAME with the drive class name.
• Replace JDBC_CONNECTION_URL with the JDBC connection URL.
• Replace SOURCE_SQL_QUERY with the SQL query to be run on the source database.
• Replace YOUR_DATASET with your BigQuery dataset, and replace YOUR_TABLE_NAME with your BigQuery table name.
• Replace PATH_TO_TEMP_DIR_ON_GCS with your Cloud Storage path to the temp directory.
• Replace CONNECTION_PROPERTIES with the JDBC connection properties if required.
• Replace CONNECTION_USERNAME with the JDBC connection username.
• Replace CONNECTION_PASSWORD with the JDBC connection password.
• Replace KMS_ENCRYPTION_KEY with the KMS Encryption Key.

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Jdbc_to_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "driverJars": "DRIVER_PATHS",
       "driverClassName": "DRIVER_CLASS_NAME",
       "connectionURL": "JDBC_CONNECTION_URL",
       "query": "SOURCE_SQL_QUERY",
       "outputTable": "YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS",
       "connectionProperties": "CONNECTION_PROPERTIES",
       "username": "CONNECTION_USERNAME",
       "password": "CONNECTION_PASSWORD",
       "KMSEncryptionKey":"KMS_ENCRYPTION_KEY"
   },
   "environment": { "zone": "us-central1-f" }
}

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.templates;

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.teleport.io.DynamicJdbcIO;
import com.google.cloud.teleport.templates.common.JdbcConverters;
import com.google.cloud.teleport.util.KMSEncryptedNestedValueProvider;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.TableRowJsonCoder;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;

/**
 * A template that copies data from a relational database using JDBC to an existing BigQuery table.
 */
public class JdbcToBigQuery {

  private static ValueProvider<String> maybeDecrypt(
      ValueProvider<String> unencryptedValue, ValueProvider<String> kmsKey) {
    return new KMSEncryptedNestedValueProvider(unencryptedValue, kmsKey);
  }

  /**
   * Main entry point for executing the pipeline. This will run the pipeline asynchronously. If
   * blocking execution is required, use the {@link
   * JdbcToBigQuery#run(JdbcConverters.JdbcToBigQueryOptions)} method to start the pipeline and
   * invoke {@code result.waitUntilFinish()} on the {@link PipelineResult}
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) {

    // Parse the user options passed from the command-line
    JdbcConverters.JdbcToBigQueryOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(JdbcConverters.JdbcToBigQueryOptions.class);

    run(options);
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  private static PipelineResult run(JdbcConverters.JdbcToBigQueryOptions options) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps: 1) Read records via JDBC and convert to TableRow via RowMapper
     *        2) Append TableRow to BigQuery via BigQueryIO
     */
    pipeline
        /*
         * Step 1: Read records via JDBC and convert to TableRow
         *         via {@link org.apache.beam.sdk.io.jdbc.JdbcIO.RowMapper}
         */
        .apply(
            "Read from JdbcIO",
            DynamicJdbcIO.<TableRow>read()
                .withDataSourceConfiguration(
                    DynamicJdbcIO.DynamicDataSourceConfiguration.create(
                            options.getDriverClassName(),
                            maybeDecrypt(options.getConnectionURL(), options.getKMSEncryptionKey()))
                        .withUsername(
                            maybeDecrypt(options.getUsername(), options.getKMSEncryptionKey()))
                        .withPassword(
                            maybeDecrypt(options.getPassword(), options.getKMSEncryptionKey()))
                        .withDriverJars(options.getDriverJars())
                        .withConnectionProperties(options.getConnectionProperties()))
                .withQuery(options.getQuery())
                .withCoder(TableRowJsonCoder.of())
                .withRowMapper(JdbcConverters.getResultSetToTableRow()))
        /*
         * Step 2: Append TableRow to an existing BigQuery table
         */
        .apply(
            "Write to BigQuery",
            BigQueryIO.writeTableRows()
                .withoutValidation()
                .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                .withCustomGcsTempLocation(options.getBigQueryLoadingTemporaryDirectory())
                .to(options.getOutputTable()));

    // Execute the pipeline and return the result.
    return pipeline.run();
  }
}