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:

BigQuery to Cloud Storage TFRecords
BigQuery to Cloud Storage Parquet (Storage API)
Cloud Bigtable to Cloud Storage Avro
Cloud Bigtable to Cloud Storage Parquet
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 Parquet to Cloud Bigtable
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
Apache Cassandra to Cloud Bigtable
Apache Hive to BigQuery
File Format Conversion

BigQuery to Cloud Storage TFRecords

The BigQuery to Cloud Storage TFRecords template is a pipeline that reads data from a BigQuery query and writes it to a Cloud Storage bucket in TFRecord format. You can specify the training, testing, and validation percentage splits. By default, the split is 1 or 100% for the training set and 0 or 0% for testing and validation sets. Note that when setting the dataset split, the sum of training, testing, and validation needs to add up to 1 or 100% (for example, 0.6+0.2+0.2). Dataflow automatically determines the optimal number of shards for each output dataset.

Requirements for this pipeline:

The BigQuery dataset and table must exist.
The output Cloud Storage bucket must exist prior to pipeline execution. Note that training, testing, and validation subdirectories do not need to preexist and will be autogenerated.

Template parameters

Parameter	Description
`readQuery`	A BigQuery SQL query that extracts data from the source. For example, `select * from dataset1.sample_table`.
`outputDirectory`	The top-level Cloud Storage path prefix at which to write the training, testing, and validation TFRecord files. For example, `gs://mybucket/output`. Subdirectories for resulting training, testing and validation TFRecord files will be automatically generated based off of outputDirectory. For example, `gs://mybucket/output/train`
`trainingPercentage`	(Optional) The percentage of query data allocated to training TFRecord files. The default value is 1, or 100%.
`testingPercentage`	(Optional) The percentage of query data allocated to testing TFRecord files. The default value is 0, or 0%.
`validationPercentage`	(Optional) The percentage of query data allocated to validation TFRecord files. The default value is 0, or 0%.
`outputSuffix`	(Optional) The file suffix for the training, testing, and validation TFRecord files that are written. The default value is `.tfrecord`.

Executing the BigQuery to Cloud Storage TFRecord files template

Console

Execute from the Google Cloud Console

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
Click Run Job.

gcloud

Execute from the gcloud command-line tool

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

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

gs://dataflow-templates/VERSION/Cloud_BigQuery_to_GCS_TensorFlow_Records

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Cloud_BigQuery_to_GCS_TensorFlow_Records \
    --parameters readQuery=READ_QUERY,outputDirectory=OUTPUT_DIRECTORY,trainingPercentage=TRAINING_PERCENTAGE,testingPercentage=TESTING_PERCENTAGE,validationPercentage=VALIDATION_PERCENTAGE,outputSuffix=OUTPUT_FILENAME_SUFFIX

Replace the following values:

JOB_NAME: a job name of your choice
READ_QUERY: the BigQuery query to be executed
OUTPUT_DIRECTORY: the Cloud Storage path prefix for output datasets
TRAINING_PERCENTAGE: the decimal percentage split for the training dataset
TESTING_PERCENTAGE: the decimal percentage split for the testing dataset
VALIDATION_PERCENTAGE: the decimal percentage split for the validation dataset
OUTPUT_FILENAME_SUFFIX: the preferred output TensorFlow Record file suffix

API

Execute from the REST API

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

gs://dataflow-templates/VERSION/Cloud_BigQuery_to_GCS_TensorFlow_Records

To execute the template with the REST API , send an HTTP POST request with your project ID. This request requires authorization.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Cloud_BigQuery_to_GCS_TensorFlow_Records
{
   "jobName": "JOB_NAME",
   "parameters": {
       "readQuery":"READ_QUERY",
       "outputDirectory":"OUTPUT_DIRECTORY",
       "trainingPercentage":"TRAINING_PERCENTAGE",
       "testingPercentage":"TESTING_PERCENTAGE",
       "validationPercentage":"VALIDATION_PERCENTAGE",
       "outputSuffix":"OUTPUT_FILENAME_SUFFIX"
   },
   "environment": { "zone": "us-central1-f" }
}

Replace the following values:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
READ_QUERY: the BigQuery query to be executed
OUTPUT_DIRECTORY: the Cloud Storage path prefix for output datasets
TRAINING_PERCENTAGE: the decimal percentage split for the training dataset
TESTING_PERCENTAGE: the decimal percentage split for the testing dataset
VALIDATION_PERCENTAGE: the decimal percentage split for the validation dataset
OUTPUT_FILENAME_SUFFIX: the preferred output TensorFlow Record file suffix

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2019 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.cloud.teleport.templates.common.BigQueryConverters.BigQueryReadOptions;
import com.google.protobuf.ByteString;
import java.util.Iterator;
import java.util.Random;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.util.Utf8;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.io.FileIO;
import org.apache.beam.sdk.io.TFRecordIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.Partition;
import org.apache.beam.sdk.transforms.Reshuffle;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
import org.tensorflow.example.Example;
import org.tensorflow.example.Feature;
import org.tensorflow.example.Features;

/**
 * Dataflow template which reads BigQuery data and writes it to GCS as a set of TFRecords. The
 * source is a SQL query.
 */
public class BigQueryToTFRecord {

  /**
   * The {@link BigQueryToTFRecord#buildFeatureFromIterator(Class, Object, Feature.Builder)} method
   * handles {@link GenericData.Array} that are passed into the {@link
   * BigQueryToTFRecord#buildFeature} method creating a TensorFlow feature from the record.
   */
  private static final String TRAIN = "train/";
  private static final String TEST = "test/";
  private static final String VAL = "val/";

  private static void buildFeatureFromIterator(
      Class<?> fieldType, Object field, Feature.Builder feature) {
    ByteString byteString;
    GenericData.Array f = (GenericData.Array) field;
    if (fieldType == Long.class) {
      Iterator<Long> longIterator = f.iterator();
      while (longIterator.hasNext()) {
        Long longValue = longIterator.next();
        feature.getInt64ListBuilder().addValue(longValue);
      }
    } else if (fieldType == double.class) {
      Iterator<Double> doubleIterator = f.iterator();
      while (doubleIterator.hasNext()) {
        double doubleValue = doubleIterator.next();
        feature.getFloatListBuilder().addValue((float) doubleValue);
      }
    } else if (fieldType == String.class) {
      Iterator<Utf8> stringIterator = f.iterator();
      while (stringIterator.hasNext()) {
        String stringValue = stringIterator.next().toString();
        byteString = ByteString.copyFromUtf8(stringValue);
        feature.getBytesListBuilder().addValue(byteString);
      }
    } else if (fieldType == boolean.class) {
      Iterator<Boolean> booleanIterator = f.iterator();
      while (booleanIterator.hasNext()) {
        Boolean boolValue = booleanIterator.next();
        int boolAsInt = boolValue ? 1 : 0;
        feature.getInt64ListBuilder().addValue(boolAsInt);
      }
    }

  }

  /**
   * The {@link BigQueryToTFRecord#buildFeature} method takes in an individual field and type
   * corresponding to a column value from a SchemaAndRecord Object returned from a BigQueryIO.read()
   * step. The method builds a TensorFlow Feature based on the type of the object- ie: STRING, TIME,
   * INTEGER etc..
   */
  private static Feature buildFeature(Object field, String type) {
    Feature.Builder feature = Feature.newBuilder();
    ByteString byteString;

    switch (type) {
      case "STRING":
      case "TIME":
      case "DATE":
        if (field instanceof GenericData.Array) {
          buildFeatureFromIterator(String.class, field, feature);
        } else {
          byteString = ByteString.copyFromUtf8(field.toString());
          feature.getBytesListBuilder().addValue(byteString);
        }
        break;
      case "BYTES":
        byteString = ByteString.copyFrom((byte[]) field);
        feature.getBytesListBuilder().addValue(byteString);
        break;
      case "INTEGER":
      case "INT64":
      case "TIMESTAMP":
        if (field instanceof GenericData.Array) {
          buildFeatureFromIterator(Long.class, field, feature);
        } else {
          feature.getInt64ListBuilder().addValue((long) field);
        }
        break;
      case "FLOAT":
      case "FLOAT64":
        if (field instanceof GenericData.Array) {
          buildFeatureFromIterator(double.class, field, feature);
        } else {
          feature.getFloatListBuilder().addValue((float) (double) field);
        }
        break;
      case "BOOLEAN":
      case "BOOL":
        if (field instanceof GenericData.Array) {
          buildFeatureFromIterator(boolean.class, field, feature);
        } else {
          int boolAsInt = (boolean) field ? 1 : 0;
          feature.getInt64ListBuilder().addValue(boolAsInt);
        }
        break;
      default:
        throw new RuntimeException("Unsupported type: " + type);
    }
    return feature.build();
  }

  /**
   * The {@link BigQueryToTFRecord#record2Example(SchemaAndRecord)} method uses takes in a
   * SchemaAndRecord Object returned from a BigQueryIO.read() step and builds a TensorFlow Example
   * from the record.
   */
  @VisibleForTesting
  protected static byte[] record2Example(SchemaAndRecord schemaAndRecord) {
    Example.Builder example = Example.newBuilder();
    Features.Builder features = example.getFeaturesBuilder();
    GenericRecord record = schemaAndRecord.getRecord();
    for (TableFieldSchema field : schemaAndRecord.getTableSchema().getFields()) {
      Feature feature = buildFeature(record.get(field.getName()), field.getType());
      features.putFeature(field.getName(), feature);
    }
    return example.build().toByteArray();
  }

  /**
   * The {@link BigQueryToTFRecord#concatURI} method uses takes in a Cloud Storage URI and a
   * subdirectory name and safely concatenates them. The resulting String is used as a sink for
   * TFRecords.
   */
  private static String concatURI(String dir, String folder) {
    if (dir.endsWith("/")) {
      return dir + folder;
    } else {
      return dir + "/" + folder;
    }
  }

  /**
   * The {@link BigQueryToTFRecord#applyTrainTestValSplit} method transforms the PCollection by
   * randomly partitioning it into PCollections for each dataset.
   */
  static PCollectionList<byte[]> applyTrainTestValSplit(PCollection<byte[]> input,
      ValueProvider<Float> trainingPercentage,
      ValueProvider<Float> testingPercentage,
      ValueProvider<Float> validationPercentage,
      Random rand) {
    return input
        .apply(Partition.of(
            3,
            (Partition.PartitionFn<byte[]>) (number, numPartitions) -> {
              Float train = trainingPercentage.get();
              Float test = testingPercentage.get();
              Float validation = validationPercentage.get();
              Double d = rand.nextDouble();
              if (train + test + validation != 1) {
                throw new RuntimeException(String.format("Train %.2f, Test %.2f, Validation"
                    + " %.2f percentages must add up to 100 percent", train, test, validation));
              }
              if (d < train) {
                return 0;
              } else if (d >= train && d < train + test) {
                return 1;
              } else {
                return 2;
              }
            }));
  }

  /**
   * Run the pipeline.
   */
  public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options. This method does not wait until the
   * pipeline is finished before returning. Invoke {@code result.waitUntilFinish()} on the result
   * object to block until the pipeline is finished running if blocking programmatic execution is
   * required.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(Options options) {
    Random rand = new Random(100); // set random seed
    Pipeline pipeline = Pipeline.create(options);

    PCollection<byte[]> bigQueryToExamples =
        pipeline
            .apply(
                "RecordToExample",
                BigQueryIO.read(BigQueryToTFRecord::record2Example)
                    .fromQuery(options.getReadQuery())
                    .withCoder(ByteArrayCoder.of())
                    .withTemplateCompatibility()
                    .withoutValidation()
                    .usingStandardSql()
                    .withMethod(BigQueryIO.TypedRead.Method.DIRECT_READ)
                // Enable BigQuery Storage API
            ).apply("ReshuffleResults", Reshuffle.viaRandomKey());

    PCollectionList<byte[]> partitionedExamples = applyTrainTestValSplit(
        bigQueryToExamples,
        options.getTrainingPercentage(),
        options.getTestingPercentage(),
        options.getValidationPercentage(),
        rand);

    partitionedExamples.get(0).apply(
        "WriteTFTrainingRecord",
        FileIO.<byte[]>write()
            .via(TFRecordIO.sink())
            .to(ValueProvider.NestedValueProvider.of(
                options.getOutputDirectory(),
                dir -> concatURI(dir, TRAIN)))
            .withNumShards(0)
            .withSuffix(options.getOutputSuffix()));

    partitionedExamples.get(1).apply(
        "WriteTFTestingRecord",
        FileIO.<byte[]>write()
            .via(TFRecordIO.sink())
            .to(ValueProvider.NestedValueProvider.of(
                options.getOutputDirectory(),
                dir -> concatURI(dir, TEST)))
            .withNumShards(0)
            .withSuffix(options.getOutputSuffix()));

    partitionedExamples.get(2).apply(
        "WriteTFValidationRecord",
        FileIO.<byte[]>write()
            .via(TFRecordIO.sink())
            .to(ValueProvider.NestedValueProvider.of(
                options.getOutputDirectory(),
                dir -> concatURI(dir, VAL)))
            .withNumShards(0)
            .withSuffix(options.getOutputSuffix()));

    return pipeline.run();
  }

  /**
   * Define command line arguments.
   */
  public interface Options
      extends BigQueryReadOptions {

    @Description("The GCS directory to store output TFRecord files.")
    ValueProvider<String> getOutputDirectory();

    void setOutputDirectory(ValueProvider<String> outputDirectory);

    @Description("The output suffix for TFRecord Files")
    @Default.String(".tfrecord")
    ValueProvider<String> getOutputSuffix();

    void setOutputSuffix(ValueProvider<String> outputSuffix);

    @Description("The training percentage split for TFRecord Files")
    @Default.Float(1)
    ValueProvider<Float> getTrainingPercentage();

    void setTrainingPercentage(ValueProvider<Float> trainingPercentage);

    @Description("The testing percentage split for TFRecord Files")
    @Default.Float(0)
    ValueProvider<Float> getTestingPercentage();

    void setTestingPercentage(ValueProvider<Float> testingPercentage);

    @Description("The validation percentage split for TFRecord Files")
    @Default.Float(0)
    ValueProvider<Float> getValidationPercentage();

    void setValidationPercentage(ValueProvider<Float> validationPercentage);
  }
}

Python

Python source code is not available.

BigQuery to Cloud Storage Parquet (Storage API)

The BigQuery to Parquet template is a batch pipeline that reads data from a BigQuery table and writes it to a Cloud Storage bucket in Parquet format. This template utilizes the BigQuery Storage API to export the data.

Requirements for this pipeline:

The input BigQuery table must exist prior to running the pipeline.
The output Cloud Storage bucket must exist prior to running the pipeline.

Template parameters

Parameter	Description
`tableRef`	The BigQuery input table location. For example, `<my-project>:<my-dataset>.<my-table>`.
`bucket`	The Cloud Storage folder in which to write the Parquet files. For example, `gs://mybucket/exports`.
`numShards`	(Optional) The number of output file shards. The default value is 1.
`fields`	(Optional) A comma-separated list of fields to select from the input BigQuery table.

Running the BigQuery to Cloud Storage Parquet template

Console

Run from the Google Cloud Console

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the BigQuery to Parquet template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
Click Run Job.

gcloud

Run from the gcloud command-line tool

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

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

gs://dataflow-templates/VERSION/flex/BigQuery_To_Parquet

gcloud beta dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --template-file-gcs-location=gs://dataflow-templates/latest/flex/BigQuery_To_Parquet \
    --parameters \
tableRef=BIGQUERY_TABLE,\
bucket=OUTPUT_DIRECTORY,\
numShards=NUM_SHARDS,\
fields=FIELDS

Replace the following values:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
BIGQUERY_TABLE: your BigQuery table name
OUTPUT_DIRECTORY: your Cloud Storage folder for output files
NUM_SHARDS: the desired number of output file shards
FIELDS: the comma-separated list of fields to select from the input BigQuery table
LOCATION: the execution region, for example, us-central1

API

Run from the REST API

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

gs://dataflow-templates/VERSION/flex/BigQuery_To_Parquet

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "tableRef": "BIGQUERY_TABLE",
          "bucket": "OUTPUT_DIRECTORY",
          "numShards": "NUM_SHARDS",
          "fields": "FIELDS"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/latest/flex/BigQuery_To_Parquet",
   }
}

Replace the following values:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
BIGQUERY_TABLE: your BigQuery table name
OUTPUT_DIRECTORY: your Cloud Storage folder for output files
NUM_SHARDS: the desired number of output file shards
FIELDS: the comma-separated list of fields to select from the input BigQuery table
LOCATION: the execution region, for example, us-central1

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2019 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.v2.templates;

import com.google.api.gax.rpc.InvalidArgumentException;
import com.google.api.services.bigquery.model.TableReference;
import com.google.cloud.bigquery.storage.v1beta1.BigQueryStorageClient;
import com.google.cloud.bigquery.storage.v1beta1.ReadOptions.TableReadOptions;
import com.google.cloud.bigquery.storage.v1beta1.Storage.CreateReadSessionRequest;
import com.google.cloud.bigquery.storage.v1beta1.Storage.ReadSession;
import com.google.cloud.bigquery.storage.v1beta1.TableReferenceProto;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.AvroCoder;
import org.apache.beam.sdk.io.FileIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead.Method;
import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
import org.apache.beam.sdk.io.parquet.ParquetIO;
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.Validation.Required;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Splitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link BigQueryToParquet} pipeline exports data from a BigQuery table to Parquet file(s) in a
 * Google Cloud Storage bucket.
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>BigQuery Table exists.
 *   <li>Google Cloud Storage bucket exists.
 * </ul>
 *
 * <p><b>Example Usage</b>
 *
 * <pre>
 * # Set the pipeline vars
 * PROJECT=my-project
 * BUCKET_NAME=my-bucket
 * TABLE={$PROJECT}:my-dataset.my-table
 *
 * # Set containerization vars
 * IMAGE_NAME=my-image-name
 * TARGET_GCR_IMAGE=gcr.io/${PROJECT}/${IMAGE_NAME}
 * BASE_CONTAINER_IMAGE=my-base-container-image
 * BASE_CONTAINER_IMAGE_VERSION=my-base-container-image-version
 * APP_ROOT=/path/to/app-root
 * COMMAND_SPEC=/path/to/command-spec
 *
 * # Build and upload image
 * mvn clean package \
 * -Dimage=${TARGET_GCR_IMAGE} \
 * -Dbase-container-image=${BASE_CONTAINER_IMAGE} \
 * -Dbase-container-image.version=${BASE_CONTAINER_IMAGE_VERSION} \
 * -Dapp-root=${APP_ROOT} \
 * -Dcommand-spec=${COMMAND_SPEC}
 *
 * # Create an image spec in GCS that contains the path to the image
 * {
 *    "docker_template_spec": {
 *       "docker_image": $TARGET_GCR_IMAGE
 *     }
 *  }
 *
 * # Execute template:
 * API_ROOT_URL="https://dataflow.googleapis.com"
 * TEMPLATES_LAUNCH_API="${API_ROOT_URL}/v1b3/projects/${PROJECT}/templates:launch"
 * JOB_NAME="bigquery-to-parquet-`date +%Y%m%d-%H%M%S-%N`"
 *
 * time curl -X POST -H "Content-Type: application/json"     \
 *     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
 *     "${TEMPLATES_LAUNCH_API}"`
 *     `"?validateOnly=false"`
 *     `"&dynamicTemplate.gcsPath=${BUCKET_NAME}/path/to/image-spec"`
 *     `"&dynamicTemplate.stagingLocation=${BUCKET_NAME}/staging" \
 *     -d '
 *      {
 *       "jobName":"'$JOB_NAME'",
 *       "parameters": {
 *           "tableRef":"'$TABLE'",
 *           "bucket":"'$BUCKET_NAME/results'",
 *           "numShards":"5",
 *           "fields":"field1,field2"
 *        }
 *       }
 *      '
 * </pre>
 */
public class BigQueryToParquet {

  /* Logger for class. */
  private static final Logger LOG = LoggerFactory.getLogger(BigQueryToParquet.class);

  /** File suffix for file to be written. */
  private static final String FILE_SUFFIX = ".parquet";

  /** Factory to create BigQueryStorageClients. */
  static class BigQueryStorageClientFactory {

    /**
     * Creates BigQueryStorage client for use in extracting table schema.
     * @return BigQueryStorageClient
     */
    static BigQueryStorageClient create() {
      try {
        return BigQueryStorageClient.create();
      } catch (IOException e) {
        LOG.error("Error connecting to BigQueryStorage API: " + e.getMessage());
        throw new RuntimeException(e);
      }
    }
  }

  /** Factory to create ReadSessions. */
  static class ReadSessionFactory {

    /**
     * Creates ReadSession for schema extraction.
     *
     * @param client BigQueryStorage client used to create ReadSession.
     * @param tableString String that represents table to export from.
     * @param tableReadOptions TableReadOptions that specify any fields in the table to filter on.
     * @return session ReadSession object that contains the schema for the export.
     */
    static ReadSession create(
        BigQueryStorageClient client, String tableString, TableReadOptions tableReadOptions) {
      TableReference tableReference = BigQueryHelpers.parseTableSpec(tableString);
      String parentProjectId = "projects/" + tableReference.getProjectId();

      TableReferenceProto.TableReference storageTableRef =
          TableReferenceProto.TableReference.newBuilder()
              .setProjectId(tableReference.getProjectId())
              .setDatasetId(tableReference.getDatasetId())
              .setTableId(tableReference.getTableId())
              .build();

      CreateReadSessionRequest.Builder builder =
          CreateReadSessionRequest.newBuilder()
              .setParent(parentProjectId)
              .setReadOptions(tableReadOptions)
              .setTableReference(storageTableRef);
      try {
        return client.createReadSession(builder.build());
      } catch (InvalidArgumentException iae) {
        LOG.error("Error creating ReadSession: " + iae.getMessage());
        throw new RuntimeException(iae);
      }
    }
  }

  /**
   * The {@link BigQueryToParquetOptions} class provides the custom execution options passed by the
   * executor at the command-line.
   */
  public interface BigQueryToParquetOptions extends PipelineOptions {
    @Description("BigQuery table to export from in the form <project>:<dataset>.<table>")
    @Required
    String getTableRef();

    void setTableRef(String tableRef);

    @Description("GCS bucket to export BigQuery table data to (e.g. gs://mybucket/folder/).")
    @Required
    String getBucket();

    void setBucket(String bucket);

    @Description("Optional: Number of shards for output file.")
    @Default.Integer(0)
    Integer getNumShards();

    void setNumShards(Integer numShards);

    @Description("Optional: Comma separated list of fields to select from the table.")
    String getFields();

    void setFields(String fields);
  }

  /**
   * The {@link BigQueryToParquet#getTableSchema(ReadSession)} method gets Avro schema for table
   * using from the {@link ReadSession} object.
   *
   * @param session ReadSession that contains schema for table, filtered by fields if any.
   * @return avroSchema Avro schema for table. If fields are provided then schema will only contain
   *     those fields.
   */
  private static Schema getTableSchema(ReadSession session) {
    Schema avroSchema;

    avroSchema = new Schema.Parser().parse(session.getAvroSchema().getSchema());
    LOG.info("Schema for export is: " + avroSchema.toString());

    return avroSchema;
  }

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    BigQueryToParquetOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(BigQueryToParquetOptions.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(BigQueryToParquetOptions options) {

    // Create the pipeline.
    Pipeline pipeline = Pipeline.create(options);

    TableReadOptions.Builder builder = TableReadOptions.newBuilder();

    /* Add fields to filter export on, if any. */
    if (options.getFields() != null) {
      builder.addAllSelectedFields(Arrays.asList(options.getFields().split(",\\s*")));
    }

    TableReadOptions tableReadOptions = builder.build();
    BigQueryStorageClient client = BigQueryStorageClientFactory.create();
    ReadSession session =
        ReadSessionFactory.create(client, options.getTableRef(), tableReadOptions);

    // Extract schema from ReadSession
    Schema schema = getTableSchema(session);
    client.close();

    TypedRead<GenericRecord> readFromBQ =
        BigQueryIO.read(SchemaAndRecord::getRecord)
            .from(options.getTableRef())
            .withTemplateCompatibility()
            .withMethod(Method.DIRECT_READ)
            .withCoder(AvroCoder.of(schema));

    if (options.getFields() != null) {
      List<String> selectedFields = Splitter.on(",").splitToList(options.getFields());
      readFromBQ =
          selectedFields.isEmpty() ? readFromBQ : readFromBQ.withSelectedFields(selectedFields);
    }

    /*
     * Steps: 1) Read records from BigQuery via BigQueryIO.
     *        2) Write records to Google Cloud Storage in Parquet format.
     */
    pipeline
        /*
         * Step 1: Read records via BigQueryIO using supplied schema as a PCollection of
         *         {@link GenericRecord}.
         */
        .apply(
            "ReadFromBigQuery", readFromBQ)
        /*
         * Step 2: Write records to Google Cloud Storage as one or more Parquet files
         *         via {@link ParquetIO}.
         */
        .apply(
            "WriteToParquet",
            FileIO.<GenericRecord>write()
                .via(ParquetIO.sink(schema))
                .to(options.getBucket())
                .withNumShards(options.getNumShards())
                .withSuffix(FILE_SUFFIX));

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

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`	The Cloud Storage path where data is 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

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Cloud Bigtable to Avro template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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_Bigtable_to_GCS_Avro

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

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID: the ID of the Cloud Bigtable table to export
OUTPUT_DIRECTORY: the Cloud Storage path where data is written, for example, gs://mybucket/somefolder
FILENAME_PREFIX: the prefix of the Avro file name, for example, 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/Cloud_Bigtable_to_GCS_Avro

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

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

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID: the ID of the Cloud Bigtable table to export
OUTPUT_DIRECTORY: the Cloud Storage path where data is written, for example, gs://mybucket/somefolder
FILENAME_PREFIX: the prefix of the Avro file name, for example, output-

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 Parquet

The Cloud Bigtable to Cloud Storage Parquet template is a pipeline that reads data from a Cloud Bigtable table and writes it to a Cloud Storage bucket in Parquet 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`	The Cloud Storage path where data is written. For example, `gs://mybucket/somefolder`.
`filenamePrefix`	The prefix of the Parquet file name. For example, `output-`.
`numShards`	The number of output file shards. For example `2`.

Running the Cloud Bigtable to Cloud Storage Parquet file template

Console

Run from the Google Cloud Console

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Cloud Bigtable to Parquet template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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_Bigtable_to_GCS_Parquet

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

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID: the ID of the Cloud Bigtable table to export
OUTPUT_DIRECTORY: the Cloud Storage path where data is written, for example, gs://mybucket/somefolder
FILENAME_PREFIX: the prefix of the Parquet file name, for example, output-
NUM_SHARDS: the number of Parquet files to output, for example, 1

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_Bigtable_to_GCS_Parquet

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

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

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID: the ID of the Cloud Bigtable table to export
OUTPUT_DIRECTORY: the Cloud Storage path where data is written, for example, gs://mybucket/somefolder
FILENAME_PREFIX: the prefix of the Parquet file name, for example, output-
NUM_SHARDS: the number of Parquet files to output, for example, 1

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2019 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 static com.google.cloud.teleport.bigtable.BigtableToAvro.toByteArray;

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 java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericRecordBuilder;
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.coders.AvroCoder;
import org.apache.beam.sdk.io.FileIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.io.parquet.ParquetIO;
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;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.PCollection;

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

  /** 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. Default file prefix is set to \"output\".")
    @Default.String("output")
    ValueProvider<String> getFilenamePrefix();

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

    @Description(
        "The maximum number of output shards produced when writing. Default number is runner defined. ")
    @Default.Integer(0)
    ValueProvider<Integer> getNumShards();

    @SuppressWarnings("unused")
    void setNumShards(ValueProvider<Integer> numShards);
  }

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line 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();
    }
  }

  /**
   * Runs a pipeline to export data from a Cloud Bigtable table to Parquet file(s) in GCS.
   *
   * @param options arguments to the pipeline
   */
  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();
    }

    /**
     * Steps:
     * 1) Read records from Bigtable.
     * 2) Convert a Bigtable Row to a GenericRecord.
     * 3) Write GenericRecord(s) to GCS in parquet format.
     */
    pipeline
        .apply("Read from Bigtable", read)
        .apply("Transform to Parquet", MapElements.via(new BigtableToParquetFn()))
        .setCoder(AvroCoder.of(GenericRecord.class, BigtableRow.getClassSchema()))
        .apply(
            "Write to Parquet in GCS",
            FileIO.<GenericRecord>write()
                .via(ParquetIO.sink(BigtableRow.getClassSchema()))
                .to(options.getOutputDirectory())
                .withPrefix(options.getFilenamePrefix())
                .withSuffix(".parquet")
                .withNumShards(options.getNumShards()));

    return pipeline.run();
  }

  /**
   * Translates a {@link PCollection} of Bigtable {@link Row} to a {@link PCollection} of {@link
   * GenericRecord}.
   */
  static class BigtableToParquetFn extends SimpleFunction<Row, GenericRecord> {
    @Override
    public GenericRecord 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 GenericRecordBuilder(BigtableRow.getClassSchema())
          .set("key", key)
          .set("cells", cells)
          .build();
    }
  }
}

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`	The Cloud Storage path where data is 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

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Cloud Bigtable to SequenceFile template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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_Bigtable_to_GCS_SequenceFile

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

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID]: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID]: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID]: the ID of the Cloud Bigtable table to export
APPLICATION_PROFILE_ID]: the ID of the Cloud Bigtable application profile to be used for the export
DESTINATION_PATH]: the Cloud Storage path where data is written, for example, gs://mybucket/somefolder
FILENAME_PREFIX]: the prefix of the SequenceFile file name, for example, 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/Cloud_Bigtable_to_GCS_SequenceFile

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Cloud_Bigtable_to_GCS_SequenceFile
{
   "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" }
}

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID]: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID]: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID]: the ID of the Cloud Bigtable table to export
APPLICATION_PROFILE_ID]: the ID of the Cloud Bigtable application profile to be used for the export
DESTINATION_PATH]: the Cloud Storage path where data is written, for example, gs://mybucket/somefolder
FILENAME_PREFIX]: the prefix of the SequenceFile file name, for example, output-

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

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`	The name of the JavaScript function to be called. For example, if your JavaScript function is `function myTransform(inJson) { ...dostuff...}`, 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 is written. For example, `gs://mybucket/somefolder/`.

Running the Datastore to Cloud Storage Text template

Console

Run from the Google Cloud Console

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Datastore to Cloud Storage Text template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Datastore_to_GCS_Text \
    --parameters \
datastoreReadGqlQuery="SELECT * FROM DATASTORE_KIND",\
datastoreReadProjectId=PROJECT_ID,\
datastoreReadNamespace=DATASTORE_NAMESPACE,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
textWritePrefix=gs://BUCKET_NAME/output/

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
BUCKET_NAME: the name of your Cloud Storage bucket
DATASTORE_KIND: your the type of your Datastore entities
DATASTORE_NAMESPACE: the namespace of your Datastore entities
JAVASCRIPT_FUNCTION: your JavaScript function name
PATH_TO_JAVASCRIPT_UDF_FILE: the Cloud Storage path to the .js file containing your JavaScript code

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

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

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

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
BUCKET_NAME: the name of your Cloud Storage bucket
DATASTORE_KIND: your the type of your Datastore entities
DATASTORE_NAMESPACE: the namespace of your Datastore entities
JAVASCRIPT_FUNCTION: your JavaScript function name
PATH_TO_JAVASCRIPT_UDF_FILE: the Cloud Storage path to the .js file containing your JavaScript code

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 IAM roles necessary to run Dataflow jobs, you must also have the appropriate 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.
`snapshotTime`	(Optional) The timestamp that corresponds to the version of the Cloud Spanner database that you want to read. The timestamp must be specified as per RFC 3339 UTC "Zulu" format. For example, `1990-12-31T23:59:60Z`. The timestamp must be in the past and Maximum timestamp staleness applies.
`spannerProjectId`	(Optional) The Google Cloud Project ID of the Cloud Spanner database that you want to read data from.

Running the Cloud Spanner to Avro Files on Cloud Storage template

Console

Run from the Google Cloud Console

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Spanner to Cloud Storage Avro template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
- 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.
Enter your parameter values in the provided parameter fields.
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

gcloud dataflow jobs run JOB_NAME \
    --gcs-location='gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro' \
    --region=DATAFLOW_REGION \
    --parameters='instanceId=INSTANCE_ID,databaseId=DATABASE_ID,outputDir=GCS_DIRECTORY

Replace the following:

JOB_NAME: a job name of your choice
DATAFLOW_REGION: the region where you want the Dataflow job to run (such as us-central1)
GCS_STAGING_LOCATION: the path for writing temporary files. For example, gs://mybucket/temp.
INSTANCE_ID: your Cloud Spanner instance ID
DATABASE_ID: your Cloud Spanner database ID
GCS_DIRECTORY: the Cloud Storage path that the Avro files are exported to

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/DATAFLOW_REGION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro
{
   "jobName": "JOB_NAME",
   "parameters": {
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "outputDir": "gs://GCS_DIRECTORY"
   }
}

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. Replace the following:

PROJECT_ID: your project ID
DATAFLOW_REGION: the region where you want the Dataflow job to run (such as us-central1)
INSTANCE_ID: your Cloud Spanner instance ID
DATABASE_ID: your Cloud Spanner database ID
GCS_DIRECTORY: the Cloud Storage path that the Avro files are exported to
JOB_NAME: a job name of your choice
- The job name must match the format cloud-spanner-export-INSTANCE_ID-DATABASE_ID to show up in the Cloud Spanner portion of the Cloud Console.

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);

    @Description("If set, specifies the time when the snapshot must be taken."
      + " String is in the RFC 3339 format. "
      + " Example - 1990-12-31T23:59:60Z"
      + " Timestamp must be in the past and Maximum timestamp staleness applies."
      + " https://cloud.google.com/spanner/docs/timestamp-bounds#maximum_timestamp_staleness")
    @Default.String(value = "")
    ValueProvider<String> getSnapshotTime();
    void setSnapshotTime(ValueProvider<String> value);

    @Description("GCP Project Id of where the Spanner table lives.")
    ValueProvider<String> getSpannerProjectId();
    void setSpannerProjectId(ValueProvider<String> value);

    @Description("If true, Timestamps are exported with timestamp-micros logical type."
      + "By default,Timestamps are exported as ISO8601 strings at nanosecond precision.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getShouldExportTimestampAsLogicalType();
    void setShouldExportTimestampAsLogicalType(ValueProvider<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()
            .withProjectId(options.getSpannerProjectId())
            .withHost(options.getSpannerHost())
            .withInstanceId(options.getInstanceId())
            .withDatabaseId(options.getDatabaseId());
    p.begin()
        .apply(
            "Run Export",
            new ExportTransform(spannerConfig, options.getOutputDir(),
                options.getTestJobId(), options.getSnapshotTime(),
                options.getShouldExportTimestampAsLogicalType()));
    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

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 Spanner table must exist prior to running the pipeline.

Template parameters

Parameter	Description
`spannerProjectId`	The Google Cloud Project ID of the Cloud Spanner database that you want to read data from.
`spannerDatabaseId`	The database ID of the requested table.
`spannerInstanceId`	The instance ID of the requested table.
`spannerTable`	The table to read the data from.
`textWritePrefix`	The directory where output text files are written. Add / at the end. For example, `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) The name of the JavaScript function to be called. For example, if your JavaScript function is `function myTransform(inJson) { ...dostuff...}`, 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

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Cloud Spanner to Cloud Storage Text template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Spanner_to_GCS_Text \
    --parameters \
spannerProjectId=PROJECT_ID,\
spannerDatabaseId=DATABASE_ID,\
spannerInstanceId=INSTANCE_ID,\
spannerTable=TABLE_ID,\
textWritePrefix=gs://BUCKET_NAME/output/,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
DATABASE_ID: the Spanner database ID
BUCKET_NAME: the name of your Cloud Storage bucket
INSTANCE_ID: the Spanner instance ID
TABLE_ID: the Spanner table ID
PATH_TO_JAVASCRIPT_UDF_FILE: the Cloud Storage path to the .js file containing your JavaScript code
JAVASCRIPT_FUNCTION: your JavaScript function 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/Spanner_to_GCS_Text

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Spanner_to_GCS_Text
{
   "jobName": "JOB_NAME",
   "parameters": {
       "spannerProjectId": "PROJECT_ID",
       "spannerDatabaseId": "DATABASE_ID",
       "spannerInstanceId": "INSTANCE_ID",
       "spannerTable": "TABLE_ID",
       "textWritePrefix": "gs://BUCKET_NAME/output/",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION"
   },
   "environment": { "zone": "us-central1-f" }
}

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
DATABASE_ID: the Spanner database ID
BUCKET_NAME: the name of your Cloud Storage bucket
INSTANCE_ID: the Spanner instance ID
TABLE_ID: the Spanner table ID
PATH_TO_JAVASCRIPT_UDF_FILE: the Cloud Storage path to the .js file containing your JavaScript code
JAVASCRIPT_FUNCTION: your JavaScript function name

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.LocalSpannerIO;
import org.apache.beam.sdk.io.gcp.spanner.ReadOperation;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
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()
            .withHost(options.getSpannerHost())
            .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", LocalSpannerIO.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`	The 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

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Cloud Storage Avro to Cloud Bigtable from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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_Bigtable

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

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID: the ID of the Cloud Bigtable table to export
INPUT_FILE_PATTERN: the Cloud Storage path pattern where data is located, for example, gs://mybucket/somefolder/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/GCS_Avro_to_Cloud_Bigtable

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

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

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID: the ID of the Cloud Bigtable table to export
INPUT_FILE_PATTERN: the Cloud Storage path pattern where data is located, for example, gs://mybucket/somefolder/prefix*

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.base.MoreObjects;
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.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 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 {
  private static final Logger LOG = LoggerFactory.getLogger(AvroToBigtable.class);

  /** Maximum number of mutations allowed per row by Cloud bigtable. */
  private static final int MAX_MUTATIONS_PER_ROW = 100000;

  private static final Boolean DEFAULT_SPLIT_LARGE_ROWS = false;

  /** 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("If true, a large row is split into multiple MutateRows requests.")
    ValueProvider<Boolean> getSplitLargeRows();

    @Description(
        "Set the option to split a large row into multiple MutateRows requests. When a row is"
            + " split across requests, updates are not atomic. ")
    void setSplitLargeRows(ValueProvider<Boolean> splitLargeRows);

    @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",
            ParDo.of(
                AvroToBigtableFn.createWithSplitLargeRows(
                    options.getSplitLargeRows(), MAX_MUTATIONS_PER_ROW)))
        .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 DoFn<BigtableRow, KV<ByteString, Iterable<Mutation>>> {
    private final ValueProvider<Boolean> splitLargeRowsFlag;
    private Boolean splitLargeRows;
    private final int maxMutationsPerRow;

    public static AvroToBigtableFn create() {
      return new AvroToBigtableFn(StaticValueProvider.of(false), MAX_MUTATIONS_PER_ROW);
    }

    public static AvroToBigtableFn createWithSplitLargeRows(
        ValueProvider<Boolean> splitLargeRowsFlag, int maxMutationsPerRequest) {
      return new AvroToBigtableFn(splitLargeRowsFlag, maxMutationsPerRequest);
    }

    private AvroToBigtableFn(
        ValueProvider<Boolean> splitLargeRowsFlag, int maxMutationsPerRequest) {
      this.splitLargeRowsFlag = splitLargeRowsFlag;
      this.maxMutationsPerRow = maxMutationsPerRequest;
    }

    @Setup
    public void setup() {
      if (splitLargeRowsFlag != null) {
        splitLargeRows = splitLargeRowsFlag.get();
      }
      splitLargeRows = MoreObjects.firstNonNull(splitLargeRows, DEFAULT_SPLIT_LARGE_ROWS);
      LOG.info("splitLargeRows set to: " + splitLargeRows);
    }

    @ProcessElement
    public void processElement(
        @Element BigtableRow row, OutputReceiver<KV<ByteString, Iterable<Mutation>>> out) {
      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();
      int cellsProcessed = 0;
      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());
        cellsProcessed++;

        if (this.splitLargeRows && cellsProcessed % maxMutationsPerRow == 0) {
          // Send a MutateRow request when we have accumulated max mutations per row.
          out.output(KV.of(key, mutations.build()));
          mutations = ImmutableList.builder();
        }
      }

      // Flush any remaining mutations.
      ImmutableList remainingMutations = mutations.build();
      if (!remainingMutations.isEmpty()) {
        out.output(KV.of(key, remainingMutations));
      }
    }
  }

  /** Copies the content in {@code byteBuffer} into a {@link ByteString}. */
  protected 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 exported from Cloud Spanner stored in 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.

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 are imported from.

Running the Cloud Storage Avro to Cloud Spanner template

Console

Run from the Google Cloud Console

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Cloud Storage Avro to Spanner template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
- 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.
Enter your parameter values in the provided parameter fields.
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

gcloud dataflow jobs run JOB_NAME \
    --gcs-location='gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner' \
    --region=DATAFLOW_REGION \
    --staging-location=GCS_STAGING_LOCATION \
    --parameters='instanceId=INSTANCE_ID,databaseId=DATABASE_ID,inputDir=GCS_DIRECTORY'

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. Replace the following:

(API only)PROJECT_ID: your project ID
DATAFLOW_REGION: the region where you want the Dataflow job to run (such as us-central1)
JOB_NAME: a job name of your choice
INSTANCE_ID: the ID of the Spanner instance that contains the database
DATABASE_ID: the ID of the Spanner database to import to
(gcloud only)GCS_STAGING_LOCATION: the path for writing temporary files, for example, gs://mybucket/temp
GCS_DIRECTORY: the Cloud Storage path where the Avro files are imported from, for example, gs://mybucket/somefolder

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/DATAFLOW_REGION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner
{
   "jobName": "JOB_NAME",
   "parameters": {
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "inputDir": "gs://GCS_DIRECTORY"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}

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. Replace the following:

(API only)PROJECT_ID: your project ID
DATAFLOW_REGION: the region where you want the Dataflow job to run (such as us-central1)
JOB_NAME: a job name of your choice
INSTANCE_ID: the ID of the Spanner instance that contains the database
DATABASE_ID: the ID of the Spanner database to import to
(gcloud only)GCS_STAGING_LOCATION: the path for writing temporary files, for example, gs://mybucket/temp
GCS_DIRECTORY: the Cloud Storage path where the Avro files are imported from, for example, gs://mybucket/somefolder

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("By default the import pipeline is not blocked on index creation, and it "
      + "may complete with indexes still being created in the background. In testing, it may "
      + "be useful to set this option to false so that the pipeline waits until indexes are "
      + "finished.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getWaitForIndexes();

    void setWaitForIndexes(ValueProvider<Boolean> value);

    @Description("By default the import pipeline is not blocked on foreign key creation, and it "
      + "may complete with foreign keys still being created in the background. In testing, it may "
      + "be useful to set this option to false so that the pipeline waits until foreign keys are "
      + "finished.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getWaitForForeignKeys();

    void setWaitForForeignKeys(ValueProvider<Boolean> value);

    @Description("Indexes and Foreign keys are created after dataload. If there are more than "
      + "40 DDL statements to be executed after dataload, it is preferable to create the "
      + "indexes before datalod. This is the flag to turn the feature off.")
    @Default.Boolean(true)
    ValueProvider<Boolean> getEarlyIndexCreateFlag();

    void setEarlyIndexCreateFlag(ValueProvider<Boolean> value);

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

    @Description("GCP Project Id of where the Spanner table lives.")
    ValueProvider<String> getSpannerProjectId();

    void setSpannerProjectId(ValueProvider<String> value);

    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()
            .withProjectId(options.getSpannerProjectId())
            .withHost(options.getSpannerHost())
            .withInstanceId(options.getInstanceId())
            .withDatabaseId(options.getDatabaseId());

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

    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 Parquet to Cloud Bigtable

The Cloud Storage Parquet to Cloud Bigtable template is a pipeline that reads data from Parquet 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 Parquet files.
The input Parquet files must exist in a Cloud Storage bucket prior to running the pipeline.
Cloud Bigtable expects a specific schema from the input Parquet 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`	The Cloud Storage path pattern where data is located. For example, `gs://mybucket/somefolder/prefix*`.

Running the Cloud Storage Parquet file to Cloud Bigtable template

Console

Run from the Google Cloud Console

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Cloud Storage Avro to Spanner template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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_Parquet_to_Cloud_Bigtable

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

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID: the ID of the Cloud Bigtable table to export
INPUT_FILE_PATTERN: the Cloud Storage path pattern where data is located, for example, gs://mybucket/somefolder/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/GCS_Parquet_to_Cloud_Bigtable

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

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

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID: the ID of the Cloud Bigtable table to export
INPUT_FILE_PATTERN: the Cloud Storage path pattern where data is located, for example, gs://mybucket/somefolder/prefix*

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2019 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 static com.google.cloud.teleport.bigtable.AvroToBigtable.toByteString;

import com.google.bigtable.v2.Mutation;
import com.google.protobuf.ByteString;
import java.nio.ByteBuffer;
import java.util.List;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.beam.runners.direct.DirectRunner;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.io.parquet.ParquetIO;
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.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link ParquetToBigtable} pipeline imports data from Parquet 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 Parquet files has a 'family' of "f1", the
 * Bigtable table should have a column family of "f1".
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>Bigtable instance.
 *   <li>Bigtable table with compatible table schema.
 *   <li>Google Cloud Storage input bucket and parquet file(s) exists.
 * </ul>
 *
 * <p><b>Example Usage</b>
 *
 * <pre>
 *
 * # Set the pipeline vars
 * PROJECT_ID=PROJECT ID HERE
 * PIPELINE_FOLDER=gs://${PROJECT_ID}/dataflow/pipelines/parquet-to-bigtable
 * BIGTABLE_INSTANCE_ID=BIGTABLE INSTANCE ID HERE
 * BIGTABLE_TABLE_ID=BIGTABLE TABLE ID HERE
 *
 * # Set the runner
 * RUNNER=DataflowRunner
 *
 * # Build the template
 * mvn compile exec:java \
 * -Dexec.mainClass=com.google.cloud.teleport.bigtable.ParquetToBigtable \
 * -Dexec.cleanupDaemonThreads=false \
 * -Dexec.args=" \
 * --project=${PROJECT_ID} \
 * --stagingLocation=${PIPELINE_FOLDER}/staging \
 * --tempLocation=${PIPELINE_FOLDER}/temp \
 * --templateLocation=${PIPELINE_FOLDER}/template \
 * --runner=${RUNNER}"
 *
 * # Execute the template
 * JOB_NAME=parquet-to-bigtable-$USER-`date +"%Y%m%d-%H%M%S%z"`
 *
 * gcloud dataflow jobs run ${JOB_NAME} \
 * --gcs-location=${PIPELINE_FOLDER}/template \
 * --zone=us-east1-d \
 * --parameters \
 * "bigtableProjectId=${PROJECT_ID},\
 * bigtableInstanceId=${BIGTABLE_INSTANCE_ID},\
 * bigtableTableId=${BIGTABLE_TABLE_ID},\
 * inputFilePattern=${PIPELINE_FOLDER}/path/to/file/filename-*.parquet"
 * </pre>
 */
public class ParquetToBigtable {
  private static final Logger LOG = LoggerFactory.getLogger(AvroToBigtable.class);

  /** Maximum number of mutations allowed per row by Cloud bigtable. */
  private static final int MAX_MUTATIONS_PER_ROW = 100000;

  private static final Boolean DEFAULT_SPLIT_LARGE_ROWS = false;

  /**
   * 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*.parquet)")
    ValueProvider<String> getInputFilePattern();

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

    @Description(
        "If true, a large row is split into multiple MutateRows requests. When a row is"
            + " split across requests, updates are not atomic. ")
    ValueProvider<Boolean> getSplitLargeRows();

    void setSplitLargeRows(ValueProvider<Boolean> splitLargeRows);
  }

  /**
   * Runs a pipeline to import Parquet 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);

  }

  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());

    /**
     * Steps: 1) Read records from Parquet File. 2) Convert a GenericRecord to a
     * KV<ByteString,Iterable<Mutation>>. 3) Write KV to Bigtable's table.
     */
    pipeline
        .apply(
            "Read from Parquet",
            ParquetIO.read(BigtableRow.getClassSchema()).from(options.getInputFilePattern()))
        .apply(
            "Transform to Bigtable",
            ParDo.of(
                ParquetToBigtableFn.createWithSplitLargeRows(
                    options.getSplitLargeRows(), MAX_MUTATIONS_PER_ROW)))
        .apply("Write to Bigtable", write);

    return pipeline.run();
  }

  static class ParquetToBigtableFn
          extends DoFn<GenericRecord, KV<ByteString, Iterable<Mutation>>> {

    private final ValueProvider<Boolean> splitLargeRowsFlag;
    private Boolean splitLargeRows;
    private final int maxMutationsPerRow;

    public static ParquetToBigtableFn create() {
      return new ParquetToBigtableFn(StaticValueProvider.of(false), MAX_MUTATIONS_PER_ROW);
    }

    public static ParquetToBigtableFn createWithSplitLargeRows(
        ValueProvider<Boolean> splitLargeRowsFlag, int maxMutationsPerRequest) {
      return new ParquetToBigtableFn(splitLargeRowsFlag, maxMutationsPerRequest);
    }

    @Setup
    public void setup() {
      if (splitLargeRowsFlag != null) {
        splitLargeRows = splitLargeRowsFlag.get();
      }
      splitLargeRows = MoreObjects.firstNonNull(splitLargeRows, DEFAULT_SPLIT_LARGE_ROWS);
      LOG.info("splitLargeRows set to: " + splitLargeRows);
    }

    private ParquetToBigtableFn(
        ValueProvider<Boolean> splitLargeRowsFlag, int maxMutationsPerRequest) {
      this.splitLargeRowsFlag = splitLargeRowsFlag;
      this.maxMutationsPerRow = maxMutationsPerRequest;
    }

    @ProcessElement
    public void processElement(ProcessContext ctx) {
      Class runner = ctx.getPipelineOptions().getRunner();
      ByteString key = toByteString((ByteBuffer) ctx.element().get(0));

      // 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();
      List<Object> cells = (List) ctx.element().get(1);
      int cellsProcessed = 0;
      for (Object element : cells) {
        Mutation.SetCell setCell = null;
        if (runner.isAssignableFrom(DirectRunner.class)) {
          setCell =
                  Mutation.SetCell.newBuilder()
                          .setFamilyName(((GenericData.Record) element).get(0).toString())
                          .setColumnQualifier(toByteString((ByteBuffer) ((GenericData.Record) element).get(1)))
                          .setTimestampMicros((Long) ((GenericData.Record) element).get(2))
                          .setValue(toByteString((ByteBuffer) ((GenericData.Record) element).get(3)))
                          .build();
        } else {
          BigtableCell bigtableCell = (BigtableCell) element;
          setCell =
                  Mutation.SetCell.newBuilder()
                          .setFamilyName(bigtableCell.getFamily().toString())
                          .setColumnQualifier(toByteString(bigtableCell.getQualifier()))
                          .setTimestampMicros(bigtableCell.getTimestamp())
                          .setValue(toByteString(bigtableCell.getValue()))
                          .build();
        }
        mutations.add(Mutation.newBuilder().setSetCell(setCell).build());
        cellsProcessed++;

        if (this.splitLargeRows && cellsProcessed % maxMutationsPerRow == 0) {
          // Send a MutateRow request when we have accumulated max mutations per row.
          ctx.output(KV.of(key, mutations.build()));
          mutations = ImmutableList.builder();
        }
      }

      // Flush any remaining mutations.
      ImmutableList remainingMutations = mutations.build();
      if (!remainingMutations.isEmpty()) {
        ctx.output(KV.of(key, remainingMutations));
      }
    }
  }
}

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`	The 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

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the SequenceFile Files on Cloud Storage to Cloud Bigtable template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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_SequenceFile_to_Cloud_Bigtable

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

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID: the ID of the Cloud Bigtable table to export
APPLICATION_PROFILE_ID: the ID of the Cloud Bigtable application profile to be used for the export
SOURCE_PATTERN: the Cloud Storage path pattern where data is located, for example, gs://mybucket/somefolder/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/GCS_SequenceFile_to_Cloud_Bigtable

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_SequenceFile_to_Cloud_Bigtable
{
   "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" }
}

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. Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PROJECT_ID: the ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from
INSTANCE_ID: the ID of the Cloud Bigtable instance that contains the table
TABLE_ID: the ID of the Cloud Bigtable table to export
APPLICATION_PROFILE_ID: the ID of the Cloud Bigtable application profile to be used for the export
SOURCE_PATTERN: the Cloud Storage path pattern where data is located, for example, gs://mybucket/somefolder/prefix*

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

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 data is appended to the destination table.

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 data is appended to the destination table. For example, `my-project-name:my-dataset.my-table`.
`bigQueryLoadingTemporaryDirectory`	The temporary directory for the 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

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Cloud Storage Text to BigQuery template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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

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

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
JAVASCRIPT_FUNCTION: the name of your UDF
PATH_TO_BIGQUERY_SCHEMA_JSON: the Cloud Storage path to the JSON file containing the schema definition
PATH_TO_JAVASCRIPT_UDF_FILE: the Cloud Storage path to the .js file containing your JavaScript code
PATH_TO_TEXT_DATA: your Cloud Storage path to your text dataset
BIGQUERY_TABLE: your BigQuery table name
PATH_TO_TEMP_DIR_ON_GCS: your Cloud Storage path to the temp 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_Text_to_BigQuery

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

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

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
JAVASCRIPT_FUNCTION: the name of your UDF
PATH_TO_BIGQUERY_SCHEMA_JSON: the Cloud Storage path to the JSON file containing the schema definition
PATH_TO_JAVASCRIPT_UDF_FILE: the Cloud Storage path to the .js file containing your JavaScript code
PATH_TO_TEXT_DATA: your Cloud Storage path to your text dataset
BIGQUERY_TABLE: your BigQuery table name
PATH_TO_TEMP_DIR_ON_GCS: your Cloud Storage path to the temp 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.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_APPEND)
                .withCustomGcsTempLocation(options.getBigQueryLoadingTemporaryDirectory()));

    pipeline.run();
  }
}

Cloud Storage Text to Datastore

The Cloud Storage Text to Datastore template is a batch pipeline that reads from text files stored in Cloud Storage and writes JSON encoded Entities to Datastore. Each line in the input text files must be in the specified JSON format.

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

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Cloud Storage Text to Datastore template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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

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=JAVASCRIPT_FUNCTION,\
datastoreWriteProjectId=PROJECT_ID,\
errorWritePath=ERROR_FILE_WRITE_PATH

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PATH_TO_INPUT_TEXT_FILES: the input files pattern on Cloud Storage
JAVASCRIPT_FUNCTION: your JavaScript function name
PATH_TO_JAVASCRIPT_UDF_FILE: the Cloud Storage path to the .js file containing your JavaScript code
ERROR_FILE_WRITE_PATH: your desired path to error file on Cloud Storage

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

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

POST https://dataflow.googleapis.com/v1b3/projects/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": "JAVASCRIPT_FUNCTION",
       "datastoreWriteProjectId": "PROJECT_ID",
       "errorWritePath": "ERROR_FILE_WRITE_PATH"
   },
   "environment": { "zone": "us-central1-f" }
}

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
PATH_TO_INPUT_TEXT_FILES: the input files pattern on Cloud Storage
JAVASCRIPT_FUNCTION: your JavaScript function name
PATH_TO_JAVASCRIPT_UDF_FILE: the Cloud Storage path to the .js file containing your JavaScript code
ERROR_FILE_WRITE_PATH: your desired path to error file on Cloud Storage

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)

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. The event time is equal to the publishing time during execution. If your pipeline relies on an accurate event time for processing, you must 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 might cause issues downstream because 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 must 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

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Cloud Storage Text to Pub/Sub (Batch) template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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

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

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
TOPIC_NAME: your Pub/Sub topic name
BUCKET_NAME: the name of your Cloud Storage bucket

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

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

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

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
TOPIC_NAME: your Pub/Sub topic name
BUCKET_NAME: the name of your Cloud Storage bucket

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 file 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, NUMERIC, 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 =
      TABLE_NAME ORDER BY ORDINAL_POSITION

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 must 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 time. 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 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 of 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 Text Files on Cloud Storage to Cloud Spanner template

Console

Run in the Google Cloud Console

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Cloud Storage Text to Cloud Spanner template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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

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

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. Replace the following:

DATAFLOW_REGION: the region where you want the Dataflow job to run (such as us-central1)
INSTANCE_ID: your Cloud Spanner instance ID
DATABASE_ID: your Cloud Spanner database ID
GCS_PATH_TO_IMPORT_MANIFEST: the Cloud Storage path to your import manifest file
JOB_NAME: a job name of your choice

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

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

{
   "jobName": "JOB_NAME",
   "parameters": {
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "importManifest": "GCS_PATH_TO_IMPORT_MANIFEST"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}

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. Replace the following:

PROJECT_ID: your project ID
DATAFLOW_REGION: the region where you want the Dataflow job to run (such as us-central1)
INSTANCE_ID: your Cloud Spanner instance ID
DATABASE_ID: your Cloud Spanner database ID
GCS_PATH_TO_IMPORT_MANIFEST: the Cloud Storage path to your import manifest file
JOB_NAME: a job name of your choice

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

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 Cloud KMS key along with a Base64-encoded username, password, and connection string parameters encrypted with the Cloud KMS key. See the Cloud 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`	The comma-separated list of driver JAR files. 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 Cloud KMS key.
`query`	The 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`	The temporary directory for the 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) The username to be used for the JDBC connection. Can be passed in as a Base64-encoded string encrypted with a Cloud KMS key.
`password`	(Optional) The password to be used for the JDBC connection. Can be passed in as a Base64-encoded string encrypted with a Cloud KMS key.
`KMSEncryptionKey`	(Optional) Cloud KMS Encryption Key to decrypt the username, password, and connection string. If Cloud 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

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the JDBC to BigQuery template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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

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=PROJECT_ID:DATASET.TABLE_NAME,
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS,\
connectionProperties=CONNECTION_PROPERTIES,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
KMSEncryptionKey=KMS_ENCRYPTION_KEY

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
DRIVER_PATHS: the comma-separated Cloud Storage path(s) of the JDBC driver(s)
DRIVER_CLASS_NAME: the drive class name
JDBC_CONNECTION_URL: the JDBC connection URL
SOURCE_SQL_QUERY: the SQL query to be run on the source database
DATASET: your BigQuery dataset, and replace TABLE_NAME: your BigQuery table name
PATH_TO_TEMP_DIR_ON_GCS: your Cloud Storage path to the temp directory
CONNECTION_PROPERTIES: the JDBC connection properties if required
CONNECTION_USERNAME: the JDBC connection username
CONNECTION_PASSWORD: the JDBC connection password
KMS_ENCRYPTION_KEY: the Cloud 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

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

POST https://dataflow.googleapis.com/v1b3/projects/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": "PROJECT_ID:DATASET.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" }
}

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
DRIVER_PATHS: the comma-separated Cloud Storage path(s) of the JDBC driver(s)
DRIVER_CLASS_NAME: the drive class name
JDBC_CONNECTION_URL: the JDBC connection URL
SOURCE_SQL_QUERY: the SQL query to be run on the source database
DATASET: your BigQuery dataset, and replace TABLE_NAME: your BigQuery table name
PATH_TO_TEMP_DIR_ON_GCS: your Cloud Storage path to the temp directory
CONNECTION_PROPERTIES: the JDBC connection properties if required
CONNECTION_USERNAME: the JDBC connection username
CONNECTION_PASSWORD: the JDBC connection password
KMS_ENCRYPTION_KEY: the Cloud KMS Encryption Key

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();
  }
}

Apache Cassandra to Cloud Bigtable

The Apache Cassandra to Cloud Bigtable template copies a table from Apache Cassandra to Cloud Bigtable. This template requires minimal configuration and replicates the table structure in Cassandra as closely as possible in Cloud Bigtable.

The Apache Cassandra to Cloud Bigtable template is useful for the following:

Migrating Apache Cassandra database when short downtime is acceptable.
Periodically replicating Cassandra tables to Cloud Bigtable for global serving.

Requirements for this pipeline:

The target Cloud Bigtable table must exist prior to running the pipeline.
Network connection between Dataflow workers and Apache Cassandra nodes.

Type conversion

The Apache Cassandra to Cloud Bigtable template automatically converts Apache Cassandra data types to Cloud Bigtable's data types.

Most primitivies are represented the same way in Cloud Bigtable and Apache Cassandra; however, the following primitives are represented differently:

Date and Timestamp are converted to DateTime objects
UUID is converted to String
Varint is converted to BigDecimal

Apache Cassandra also natively supports more complex types such as Tuple, List, Set and Map. Tuples are not supported by this pipeline as there is no corresponding type in the Apache Beam.

For example, in Apache Cassandra you can have a column of type List called "mylist" and values like those in the following table:

row	mylist
1	`(a,b,c)`

The pipeline expands the list column into three different columns (known in Cloud Bigtable as column qualifiers). The name of the columns is "mylist" but the pipeline appends the index of the item in the list, such as "mylist[0]".

row	mylist[0]	mylist[1]	mylist[2]
1	a	b	c

The pipeline handles sets the same way as lists but adds an additional suffix to denote if the cell is a key or a value.

row	mymap
1	`{"first_key":"first_value","another_key":"different_value"}`

After transformation, table appears as follows:

row	mymap[0].key	mymap[0].value	mymap[1].key	mymap[1].value
1	first_key	first_value	another_key	different_value

Primary key conversion

In Apache Cassandra, a primary key is defined using data definition language. The primary key is either simple, composite, or compound with the clustering columns. Cloud Bigtable supports manual row-key construction, ordered lexicographically on a byte array. The pipeline automatically collects information about the type of key and constructs a key based on best practices for building row-keys based on multiple values.

Template parameters

Parameter	Description
`cassandraHosts`	The hosts of the Apache Cassandra nodes in a comma-separated list.
`cassandraPort`	(Optional) The TCP port to reach Apache Cassandra on the nodes (defaults to `9042`).
`cassandraKeyspace`	The Apache Cassandra keyspace where the table is located.
`cassandraTable`	The Apache Cassandra table to be copied.
`bigtableProjectId`	The Google Cloud Project ID of the Cloud Bigtable instance where the Apache Cassandra table is copied.
`bigtableInstanceId`	The Cloud Bigtable instance ID in which to copy the Apache Cassandra table.
`bigtableTableId`	The name of the Cloud Bigtable table in which to copy the Apache Cassandra table.
`defaultColumnFamily`	(Optional) The name of the Cloud Bigtable table's column family (defaults to `default`).
`rowKeySeparator`	(Optional) The separator used to build row-key (defaults to `#`).

Running the Apache Cassandra to Cloud Bigtable template

Console

Run from the Google Cloud Console

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Apache Cassandra to Cloud Bigtable template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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/Cassandra_To_Cloud_Bigtable

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Cassandra_To_Cloud_Bigtable \
    --parameters\
bigtableProjectId=PROJECT_ID,\
bigtableInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableTableId=BIGTABLE_TABLE_ID,\
cassandraHosts=CASSANDRA_HOSTS,\
cassandraKeyspace=CASSANDRA_KEYSPACE,\
cassandraTable=CASSANDRA_TABLE

Replace the following:

PROJECT_ID: your project ID where Cloud Bigtable is located
JOB_NAME: a job name of your choice
BIGTABLE_INSTANCE_ID: the Cloud Bigtable instance id
BIGTABLE_TABLE_ID: the name of your Cloud Bigtable table name
CASSANDRA_HOSTS: the Apache Cassandra host list; if multiple hosts are provided, follow the instruction on how to escape commas
CASSANDRA_KEYSPACE: the Apache Cassandra keyspace where table is located
CASSANDRA_TABLE: the Apache Cassandra table that needs to be migrated

API

Run from the REST API

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

gs://dataflow-templates/VERSION/Cassandra_To_Cloud_Bigtable

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Cassandra_To_Cloud_Bigtable
{
   "jobName": "JOB_NAME",
   "parameters": {
       "bigtableProjectId": "PROJECT_ID",
       "bigtableInstanceId": "BIGTABLE_INSTANCE_ID",
       "bigtableTableId": "BIGTABLE_TABLE_ID",
       "cassandraHosts": "CASSANDRA_HOSTS",
       "cassandraKeyspace": "CASSANDRA_KEYSPACE",
       "cassandraTable": "CASSANDRA_TABLE"
   },
   "environment": { "zone": "us-central1-f" }
}

Replace the following:

PROJECT_ID: your project ID where Cloud Bigtable is located
JOB_NAME: a job name of your choice
BIGTABLE_INSTANCE_ID: the Cloud Bigtable instance id
BIGTABLE_TABLE_ID: the name of your Cloud Bigtable table name
CASSANDRA_HOSTS: the Apache Cassandra host list; if multiple hosts are provided, follow the instruction on how to escape commas
CASSANDRA_KEYSPACE: the Apache Cassandra keyspace where table is located
CASSANDRA_TABLE: the Apache Cassandra table that needs to be migrated

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2019 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.datastax.driver.core.Session;
import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.SerializableCoder;
import org.apache.beam.sdk.io.cassandra.CassandraIO;
import org.apache.beam.sdk.io.cassandra.Mapper;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
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;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.Row;

/**
 * This Dataflow Template performs a one off copy of one table from Apache Cassandra to Cloud
 * Bigtable. It is designed to require minimal configuration and aims to replicate the table
 * structure in Cassandra as closely as possible in Cloud Bigtable. To run the pipeline go to <link>
 * enter the required configuration and press <run>
 *
 * <p>The minimum required configuration required to run the pipeline is:
 *
 * <ul>
 *   <li><b>cassandraHosts:</b> The hosts of the Cassandra nodes in a comma separated value list.
 *   <li><b>cassandraPort:</b> The tcp port where Cassandra can be reached on the nodes.
 *   <li><b>cassandraKeyspace:</b> The Cassandra keyspace where the table is located.
 *   <li><b>cassandraTable:</b> The Cassandra table to be copied.
 *   <li><b>bigtableProjectId:</b> The Project ID of the Bigtable instance where the Cassandra table
 *       should be copied.
 *   <li><b>bigtableInstanceId:</b> The Bigtable Instance ID where the Cassandra table should be
 *       copied.
 *   <li><b>bigtableTableId:</b> The name of the Bigtable table where the Cassandra table should be
 *       copied.
 * </ul>
 */
final class CassandraToBigtable {

  public interface Options extends PipelineOptions {

    @Description("Cassandra hosts to read from")
    ValueProvider<String> getCassandraHosts();

    @SuppressWarnings("unused")
    void setCassandraHosts(ValueProvider<String> hosts);

    @Description("Cassandra port")
    @Default.Integer(9042)
    ValueProvider<Integer> getCassandraPort();

    @SuppressWarnings("unused")
    void setCassandraPort(ValueProvider<Integer> port);

    @Description("Cassandra keyspace to read from")
    ValueProvider<String> getCassandraKeyspace();

    @SuppressWarnings("unused")
    void setCassandraKeyspace(ValueProvider<String> keyspace);

    @Description("Cassandra table to read from")
    ValueProvider<String> getCassandraTable();

    @SuppressWarnings("unused")
    void setCassandraTable(ValueProvider<String> cassandraTable);

    @Description("Project where the destination Bigtable instance is located")
    ValueProvider<String> getBigtableProjectId();

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

    @Description("Target Bigtable instance id")
    ValueProvider<String> getBigtableInstanceId();

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

    @Description("Target Bigtable table")
    ValueProvider<String> getBigtableTableId();

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

    @Description("Default Column Family")
    @Default.String("default")
    ValueProvider<String> getDefaultColumnFamily();

    @SuppressWarnings("unused")
    void setDefaultColumnFamily(ValueProvider<String> defaultColumnFamily);

    @Description("RowKeySeparator")
    @Default.String("#")
    ValueProvider<String> getRowKeySeparator();

    @SuppressWarnings("unused")
    void setRowKeySeparator(ValueProvider<String> rowKeySeparator);

    @Description(
        "If true, a large row is split into multiple MutateRows requests. When a row is"
            + " split across requests, updates are not atomic. ")
    ValueProvider<Boolean> getSplitLargeRows();

    void setSplitLargeRows(ValueProvider<Boolean> splitLargeRows);
  }

  /**
   * Runs a pipeline to copy one Cassandra table to Cloud Bigtable.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {

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

    // Split the Cassandra Hosts value provider into a list value provider.
    ValueProvider.NestedValueProvider<List<String>, String> hosts =
        ValueProvider.NestedValueProvider.of(
            options.getCassandraHosts(),
            (SerializableFunction<String, List<String>>) value -> Arrays.asList(value.split(",")));

    Pipeline p = Pipeline.create(options);

    // Create a factory method to inject the CassandraRowMapperFn to allow custom type mapping.
    SerializableFunction<Session, Mapper> cassandraObjectMapperFactory =
        new CassandraRowMapperFactory(options.getCassandraTable(), options.getCassandraKeyspace());

    CassandraIO.Read<Row> source =
        CassandraIO.<Row>read()
            .withHosts(hosts)
            .withPort(options.getCassandraPort())
            .withKeyspace(options.getCassandraKeyspace())
            .withTable(options.getCassandraTable())
            .withMapperFactoryFn(cassandraObjectMapperFactory)
            .withEntity(Row.class)
            .withCoder(SerializableCoder.of(Row.class));

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

    p.apply("Read from Cassandra", source)
        .apply(
            "Convert Row",
            ParDo.of(
                BeamRowToBigtableFn.createWithSplitLargeRows(
                    options.getRowKeySeparator(),
                    options.getDefaultColumnFamily(),
                    options.getSplitLargeRows(),
                    BeamRowToBigtableFn.MAX_MUTATION_PER_REQUEST)))
        .apply("Write to Bigtable", sink);
    p.run();
  }
}

Apache Hive to BigQuery

The Apache Hive to BigQuery template is a batch pipeline that reads from an Apache Hive table and writes it to a BigQuery table.

Requirements for this pipeline:

The target Cloud Bigtable table must exist prior to running the pipeline.
Network connection must exist between Dataflow workers and Apache Hive nodes.
Network connection must exist between Dataflow workers and the Apache Thrift server node.
The BigQuery dataset must exist prior to pipeline execution.

Template parameters

Parameter	Description
`metastoreUri`	The Apache Thrift server URI such as `thrift://thrift-server-host:port`.
`hiveDatabaseName`	The Apache Hive database name that contains the table you want to export.
`hiveTableName`	The Apache Hive table name that you want to export.
`outputTableSpec`	The BigQuery output table location, in the format of `<my-project>:<my-dataset>.<my-table>`
`hivePartitionCols`	(Optional) The comma-separated list of the Apache Hive partition columns.
`filterString`	(Optional) The filter string for the input Apache Hive table.
`partitionType`	(Optional) The partition type in BigQuery. Currently, only `Time` is supported.
`partitionCol`	(Optional) The partition column name in the output BigQuery table.

Running the Apache Hive to BigQuery template

Console

Run from the Google Cloud Console

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the Apache Hive to BigQuery template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
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/Hive_To_BigQuery

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Hive_To_BigQuery \
    --parameters\
metastoreUri=METASTORE_URI,\
hiveDatabaseName=HIVE_DATABASE_NAME,\
hiveTableName=HIVE_TABLE_NAME,\
outputTableSpec=PROJECT_ID:DATASET.TABLE_NAME,\
hivePartitionCols=HIVE_PARTITION_COLS,\
filterString=FILTER_STRING,\
partitionType=PARTITION_TYPE,\
partitionCol=PARTITION_COL

Replace the following:

PROJECT_ID: your project ID where BigQuery is located
JOB_NAME: the job name of your choice
DATASET: your BigQuery dataset
TABLE_NAME: your BigQuery table name
METASTORE_URI: the Apache Thrift server URI
HIVE_DATABASE_NAME: the Apache Hive database name that contains the table you want to export
HIVE_TABLE_NAME: the Apache Hive table name that you want to export
HIVE_PARTITION_COLS: the comma-separated list of your Apache Hive partition columns
FILTER_STRING: the filter string for the Apache Hive input table
PARTITION_TYPE: the partition type in BigQuery
PARTITION_COL: the name of the BigQuery partition column

API

Run from the REST API

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

gs://dataflow-templates/VERSION/Hive_To_BigQuery

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Hive_To_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "metastoreUri": "METASTORE_URI",
       "hiveDatabaseName": "HIVE_DATABASE_NAME",
       "hiveTableName": "HIVE_TABLE_NAME",
       "outputTableSpec": "PROJECT_ID:DATASET.TABLE_NAME",
       "hivePartitionCols": "HIVE_PARTITION_COLS",
       "filterString": "FILTER_STRING",
       "partitionType": "PARTITION_TYPE",
       "partitionCol": "PARTITION_COL"
   },
   "environment": { "zone": "us-central1-f" }
}

Replace the following:

PROJECT_ID: your project ID where BigQuery is located
JOB_NAME: the job name of your choice
DATASET: your BigQuery dataset
TABLE_NAME: your BigQuery table name
METASTORE_URI: the Apache Thrift server URI
HIVE_DATABASE_NAME: the Apache Hive database name that contains the table you want to export
HIVE_TABLE_NAME: the Apache Hive table name that you want to export
HIVE_PARTITION_COLS: the comma-separated list of your Apache Hive partition columns
FILTER_STRING: the filter string for the Apache Hive input table
PARTITION_TYPE: the partition type in BigQuery
PARTITION_COL: the name of the BigQuery partition column

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2020 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.v2.templates;

import com.google.api.services.bigquery.model.TimePartitioning;
import com.google.cloud.teleport.v2.options.BigQueryCommonOptions.WriteOptions;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
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.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.io.gcp.bigquery.BigQueryUtils;
import org.apache.beam.sdk.io.hcatalog.HCatToRow;
import org.apache.beam.sdk.io.hcatalog.HCatalogIO;
import org.apache.beam.sdk.io.hdfs.HadoopFileSystemOptions;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.Row;
import org.apache.hadoop.conf.Configuration;

/**
 * The {@link HiveToBigQuery} pipeline is a batch pipeline which ingests data
 * from Hive and outputs the resulting records to BigQuery.
 *
 */

public final class HiveToBigQuery {

  /**
   * The {@link Options} class provides the custom execution options passed by the executor at the
   * command-line.
   */
  public interface Options extends DataflowPipelineOptions, HadoopFileSystemOptions, WriteOptions {
    @Description("hive metastore uri")
    @Required
    String getMetastoreUri();

    void setMetastoreUri(String value);

    @Description("Hive database name")
    @Required
    String getHiveDatabaseName();

    void setHiveDatabaseName(String value);

    @Description("Hive table name")
    @Required
    String getHiveTableName();

    void setHiveTableName(String value);

    @Description("comma separated list of the columns of Hive partition columns")
    List<java.lang.String> getHivePartitionCols();

    void setHivePartitionCols(List<java.lang.String> value);

    @Description("the filter details")
    String getFilterString();

    void setFilterString(String value);
  }

  /**
   * The main entry-point for pipeline execution. This method will start the pipeline but will not
   * wait for it's execution to finish. If blocking execution is required, use the {@link
   * HiveToBigQuery#run(Options)} method to start the pipeline and invoke
   * {@code
   * result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line args passed by the executor.
   */
  public static void main(String[] args) {
    Options options =
            PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    run(options);
  }

  /**
   * Configures the HCatalog source.
   *
   * @param options The execution options.
   * @param configProperties The configuration properties like metastore URI
   * @return hiveRead Configured HCatalogIO.Read method
   */
  private static HCatalogIO.Read getHCatalogIORead(Options options, Map<String, String> configProperties) {
    HCatalogIO.Read hiveRead = HCatalogIO.read()
        .withConfigProperties(configProperties)
        .withDatabase(options.getHiveDatabaseName())
        .withTable(options.getHiveTableName());

    if (options.getHivePartitionCols() != null) {
      hiveRead = hiveRead.withPartitionCols(options.getHivePartitionCols());
    }

    if (options.getFilterString() != null) {
      hiveRead = hiveRead.withFilter(options.getFilterString());
    }
    return hiveRead;
  }

  /**
   * Configures the HCatalog source.
   *
   * @param options The execution options.
   * @param configProperties The configuration properties like metastore URI
   * @return bqWrite Configured BigQueryIO.Write method
   */
  private static BigQueryIO.Write getBigQueryIOWrite(Options options, PCollection<Row> rows) {
    BigQueryIO.Write bqWrite = BigQueryIO.<Row>write()
            .to(options.getOutputTableSpec())
            .withSchema(BigQueryUtils.toTableSchema(rows.getSchema()))
            .withFormatFunction(BigQueryUtils.toTableRow())
            .withCreateDisposition(CreateDisposition.valueOf(options.getCreateDisposition()))
            .withWriteDisposition(WriteDisposition.valueOf(options.getWriteDisposition()))
            .withExtendedErrorInfo();

    if (options.getPartitionType() == "Time") {
      bqWrite = bqWrite.withTimePartitioning(new TimePartitioning().setField(options.getPartitionCol()));
    }
    return bqWrite;
  }

  /**
   * Runs the pipeline to completion with the specified options. This method does not wait until the
   * pipeline is finished before returning. Invoke {@code result.waitUntilFinish()} on the result
   * object to block until the pipeline is finished running if blocking programmatic execution is
   * required.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(Options options) {
    // Set Hadoop configurations
    Configuration hadoopConf = new Configuration();
    options.setHdfsConfiguration(Collections.singletonList(hadoopConf));

    Map<String, String> configProperties = new HashMap<String, String>();
    configProperties.put("hive.metastore.uris", options.getMetastoreUri());

    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    PCollection<Row> rows =
        pipeline
            /*
             * Step #1: Read hive table rows from Hive.
             */
            .apply(
                "Read from Hive source",
                    HCatToRow.fromSpec(
                        getHCatalogIORead(options, configProperties)));

    /*
     * Step #2: Write table rows out to BigQuery
     */
    rows.apply(
            "Write records to BigQuery", getBigQueryIOWrite(options, rows)
    );
    return pipeline.run();
  }
}

File Format Conversion (Avro, Parquet, CSV)

The File Format Conversion template is a batch pipeline that converts files stored on Cloud Storage from one supported format to another.

The following format conversions are supported:

CSV to Avro.
CSV to Parquet.
Avro to Parquet.
Parquet to Avro.

Requirements for this pipeline:

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

Template parameters

Parameter	Description
`inputFileFormat`	The input file format. Must be one of `[csv, avro, parquet]`.
`outputFileFormat`	The output file format. Must be one of `[avro, parquet]`.
`inputFileSpec`	The Cloud Storage path pattern for input files. For example, `gs://bucket-name/path/*.csv`
`outputBucket`	The Cloud Storage folder to write output files. This path must end with a slash. For example, `gs://bucket-name/output/`
`schema`	The Cloud Storage path to the Avro schema file. For example, `gs://bucket-name/schema/my-schema.avsc`
`containsHeaders`	(Optional) The input CSV files contain a header record (true/false). The default value is `false`. Only required when reading CSV files.
`csvFormat`	(Optional) The CSV format specification to use for parsing records. The default value is `Default`. See Apache Commons CSV Format for more details.
`delimiter`	(Optional) The field delimiter used by the input CSV files.
`outputFilePrefix`	(Optional) The output file prefix. The default value is `output`.
`numShards`	(Optional) The number of output file shards.

Running the File Format Conversion template

Console

Run from the Google Cloud Console

Go to the Dataflow page in the Cloud Console.

Go to the Dataflow page

Click Create job from template.

Select the File Format Conversion template from the Dataflow template drop-down menu.
Enter a job name in the Job Name field.
Enter your parameter values in the provided parameter fields.
Click Run Job.

gcloud

Run from the gcloud command-line tool

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

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

gs://dataflow-templates/VERSION/flex/File_Format_Conversion

gcloud beta dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --template-file-gcs-location=gs://dataflow-templates/latest/flex/File_Format_Conversion \
    --parameters \
inputFileFormat=INPUT_FORMAT,\
outputFileFormat=OUTPUT_FORMAT,\
inputFileSpec=INPUT_FILES,\
schema=SCHEMA,\
outputBucket=OUTPUT_FOLDER

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
INPUT_FORMAT: the file format of the input file; must be one of [csv, avro, parquet]
OUTPUT_FORMAT: the file format of the output files; must be one of [avro, parquet]
INPUT_FILES: the path pattern for input files
OUTPUT_FOLDER: your Cloud Storage folder for output files
SCHEMA: the path to the Avro schema file
LOCATION: the execution region, for example, us-central1

API

Run from the REST API

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

gs://dataflow-templates/VERSION/flex/File_Format_Conversion

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "inputFileFormat": "INPUT_FORMAT",
          "outputFileFormat": "OUTPUT_FORMAT",
          "inputFileSpec": "INPUT_FILES",
          "schema": "SCHEMA",
          "outputBucket": "OUTPUT_FOLDER"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/latest/flex/File_Format_Conversion",
   }
}

Replace the following:

PROJECT_ID: your project ID
JOB_NAME: a job name of your choice
INPUT_FORMAT: the file format of the input file; must be one of [csv, avro, parquet]
OUTPUT_FORMAT: the file format of the output files; must be one of [avro, parquet]
INPUT_FILES: the path pattern for input files
OUTPUT_FOLDER: your Cloud Storage folder for output files
SCHEMA: the path to the Avro schema file
LOCATION: the execution region, for example, us-central1

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2019 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.v2.templates;

import com.google.cloud.teleport.v2.transforms.AvroConverters.AvroOptions;
import com.google.cloud.teleport.v2.transforms.CsvConverters.CsvPipelineOptions;
import com.google.cloud.teleport.v2.transforms.ParquetConverters.ParquetOptions;
import java.io.IOException;
import java.util.EnumMap;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link FileFormatConversion} pipeline takes in an input file, converts it to a desired format
 * and saves it to Cloud Storage. Supported file transformations are:
 *
 * <ul>
 *   <li>Csv to Avro
 *   <li>Csv to Parquet
 *   <li>Avro to Parquet
 *   <li>Parquet to Avro
 * </ul>
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>Input file exists in Google Cloud Storage.
 *   <li>Google Cloud Storage output bucket exists.
 * </ul>
 *
 * <p><b>Example Usage</b>
 *
 * <pre>
 * # Set the pipeline vars
 * PROJECT=my-project
 * BUCKET_NAME=my-bucket
 *
 * # Set containerization vars
 * IMAGE_NAME=my-image-name
 * TARGET_GCR_IMAGE=gcr.io/${PROJECT}/${IMAGE_NAME}
 * BASE_CONTAINER_IMAGE=my-base-container-image
 * BASE_CONTAINER_IMAGE_VERSION=my-base-container-image-version
 * APP_ROOT=/path/to/app-root
 * COMMAND_SPEC=/path/to/command-spec
 *
 * # Set vars for execution
 * export INPUT_FILE_FORMAT=Csv
 * export OUTPUT_FILE_FORMAT=Avro
 * export AVRO_SCHEMA_PATH=gs://path/to/avro/schema
 * export HEADERS=false
 * export DELIMITER=","
 *
 * # Build and upload image
 * mvn clean package \
 * -Dimage=${TARGET_GCR_IMAGE} \
 * -Dbase-container-image=${BASE_CONTAINER_IMAGE} \
 * -Dbase-container-image.version=${BASE_CONTAINER_IMAGE_VERSION} \
 * -Dapp-root=${APP_ROOT} \
 * -Dcommand-spec=${COMMAND_SPEC}
 *
 * # Create an image spec in GCS that contains the path to the image
 * {
 *    "docker_template_spec": {
 *       "docker_image": $TARGET_GCR_IMAGE
 *     }
 *  }
 *
 * # Execute template:
 * API_ROOT_URL="https://dataflow.googleapis.com"
 * TEMPLATES_LAUNCH_API="${API_ROOT_URL}/v1b3/projects/${PROJECT}/templates:launch"
 * JOB_NAME="csv-to-avro-`date +%Y%m%d-%H%M%S-%N`"
 *
 * time curl -X POST -H "Content-Type: application/json"     \
 *     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
 *     "${TEMPLATES_LAUNCH_API}"`
 *     `"?validateOnly=false"`
 *     `"&dynamicTemplate.gcsPath=${BUCKET_NAME}/path/to/image-spec"`
 *     `"&dynamicTemplate.stagingLocation=${BUCKET_NAME}/staging" \
 *     -d '
 *      {
 *       "jobName":"'$JOB_NAME'",
 *       "parameters": {
 *            "inputFileFormat":"'$INPUT_FILE_FORMAT'",
 *            "outputFileFormat":"'$OUTPUT_FILE_FORMAT'",
 *            "inputFileSpec":"'$BUCKET_NAME/path/to/input-file'",
 *            "outputBucket":"'$BUCKET_NAME/path/to/output-location/'",
 *            "containsHeaders":"'$HEADERS'",
 *            "schema":"'$AVRO_SCHEMA_PATH'",
 *            "outputFilePrefix":"output-file",
 *            "numShards":"3",
 *            "delimiter":"'$DELIMITER'"
 *         }
 *       }
 *      '
 * </pre>
 */
public class FileFormatConversion {

  /** Logger for class. */
  private static final Logger LOG = LoggerFactory.getLogger(FileFormatConversionFactory.class);

  private static EnumMap<ValidFileFormats, String> validFileFormats =
      new EnumMap<ValidFileFormats, String>(ValidFileFormats.class);

  /**
   * The {@link FileFormatConversionOptions} provides the custom execution options passed by the
   * executor at the command-line.
   */
  public interface FileFormatConversionOptions
      extends PipelineOptions, CsvPipelineOptions, AvroOptions, ParquetOptions {
    @Description("Input file format.")
    @Required
    String getInputFileFormat();

    void setInputFileFormat(String inputFileFormat);

    @Description("Output file format.")
    @Required
    String getOutputFileFormat();

    void setOutputFileFormat(String outputFileFormat);
  }

  /** The {@link ValidFileFormats} enum contains all valid file formats. */
  public enum ValidFileFormats {
    CSV,
    AVRO,
    PARQUET
  }

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    FileFormatConversionOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(FileFormatConversionOptions.class);

    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options.
   *
   * @param options The execution options.
   * @return The pipeline result.
   * @throws RuntimeException thrown if incorrect file formats are passed.
   */
  public static PipelineResult run(FileFormatConversionOptions options) {
    String inputFileFormat = options.getInputFileFormat().toUpperCase();
    String outputFileFormat = options.getOutputFileFormat().toUpperCase();

    validFileFormats.put(ValidFileFormats.CSV, "CSV");
    validFileFormats.put(ValidFileFormats.AVRO, "AVRO");
    validFileFormats.put(ValidFileFormats.PARQUET, "PARQUET");

    try {
      if (inputFileFormat.equals(outputFileFormat)) {
        LOG.error("Input and output file format cannot be the same.");
        throw new IOException();
      }
      if (!validFileFormats.containsValue(inputFileFormat)
          || !validFileFormats.containsValue(outputFileFormat)) {
        LOG.error("Invalid input or output file format.");
        throw new IOException();
      }
    } catch (IOException e) {
      throw new RuntimeException("Provide correct input/output file format.");
    }

    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    pipeline.apply(
        inputFileFormat + " to " + outputFileFormat,
        FileFormatConversionFactory.FileFormat.newBuilder()
            .setOptions(options)
            .setInputFileFormat(inputFileFormat)
            .setOutputFileFormat(outputFileFormat)
            .build());

    return pipeline.run();
  }
}