Get started with Google-provided templates

Google provides a set of open source Dataflow templates. For general information about templates, see the Overview page. To get started, use the WordCount template. See other Google-provided templates:

Streaming templates

Templates for processing data continuously:

• Pub/Sub Subscription to BigQuery
• Pub/Sub Topic to BigQuery
• Pub/Sub Avro to BigQuery
• Pub/Sub Proto to BigQuery
• Pub/Sub to Pub/Sub
• Pub/Sub to Splunk
• Pub/Sub to Cloud Storage Avro
• Pub/Sub to Cloud Storage Text
• Pub/Sub to MongoDB
• Pub/Sub to Elasticsearch
• Cloud Storage Text to BigQuery (Stream)
• Cloud Spanner change streams to BigQuery
• Cloud Spanner change streams to Cloud Storage
• Cloud Storage Text to Pub/Sub (Stream)
• Data Masking/Tokenization from Cloud Storage to BigQuery (using Cloud DLP)
• Change Data Capture to BigQuery (Stream)
• Apache Kafka to BigQuery
• Datastream to BigQuery
• Datastream to PostgreSQL
• Datastream to Cloud Spanner
• Pub/Sub to Java Database Connectivity (JDBC)

Batch templates

Templates for processing data in bulk:

• BigQuery to Cloud Storage TFRecords
• BigQuery export to Parquet (via Storage API)
• BigQuery to Elasticsearch
• Bigtable to Cloud Storage Avro
• Bigtable to Cloud Storage Parquet
• Bigtable to Cloud Storage SequenceFiles
• Datastore to Cloud Storage Text [Deprecated]
• Firestore (Datastore mode) to Cloud Storage Text
• Cloud Spanner to Cloud Storage Avro
• Cloud Spanner to Cloud Storage Text
• Cloud Storage Avro to Bigtable
• Cloud Storage Avro to Cloud Spanner
• Cloud Storage Parquet to Bigtable
• Cloud Storage SequenceFiles to Bigtable
• Cloud Storage Text to BigQuery
• Cloud Storage Text to Datastore [Deprecated]
• Cloud Storage Text to Firestore (Datastore mode)
• Cloud Storage Text to Pub/Sub (Batch)
• Cloud Storage Text to Cloud Spanner
• Cloud Storage to Elasticsearch
• Java Database Connectivity (JDBC) to BigQuery
• Java Database Connectivity (JDBC) to Pub/Sub
• Apache Cassandra to Bigtable

Utility templates

• File Format Conversion
• Bulk Compress Cloud Storage Files
• Bulk Decompress Cloud Storage Files
• Datastore Bulk Delete [Deprecated]
• Firestore (Datastore mode) Bulk Delete
• Streaming Data Generator to Pub/Sub, BigQuery, and Cloud Storage

WordCount

The WordCount template is a batch pipeline that reads text from Cloud Storage, tokenizes the text lines into individual words, and performs a frequency count on each of the words. For more information about WordCount, see WordCount Example Pipeline.

If the Cloud Storage bucket is outside of your service perimeter, create an egress rule that allows access to the bucket.

Template parameters

Parameter	Description
inputFile	The Cloud Storage input file's path.
output	The Cloud Storage output file's path and prefix.

Running the WordCount template

Console

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

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

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

gcloud

Note: To use the Google Cloud CLI to run classic templates, you must have Google Cloud CLI version 138.0.0 or later.

In your shell or terminal, run the template:

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Word_Count \
    --region REGION_NAME \
    --parameters \
    inputFile=gs://dataflow-samples/shakespeare/kinglear.txt,\
    output=gs://BUCKET_NAME/output/my_output

Replace the following:

• JOB_NAME: a unique job name of your choice
• REGION_NAME: the regional endpoint where you want to deploy your Dataflow job—for example, us-central1
• BUCKET_NAME: the name of your Cloud Storage bucket

API

To run the template using the REST API, send an HTTP POST request. For more information on the API and its authorization scopes, see projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/latest/Word_Count
{
    "jobName": "JOB_NAME",
    "parameters": {
       "inputFile" : "gs://dataflow-samples/shakespeare/kinglear.txt",
       "output": "gs://BUCKET_NAME/output/my_output"
    },
    "environment": { "zone": "us-central1-f" }
}

Replace the following:

• PROJECT_ID: the Cloud project ID where you want to run the Dataflow job
• JOB_NAME: a unique job name of your choice
• LOCATION: the regional endpoint where you want to deploy your Dataflow job—for example, us-central1
• BUCKET_NAME: the name of your Cloud Storage bucket

Template source code

Java

View on GitHub Feedback

/*
 * Copyright (C) 2016 Google LLC
 *
 * 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.io.TextIO;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
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.Count;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;

/** An template that counts words in Shakespeare. */
public class WordCount {

  static class ExtractWordsFn extends DoFn<String, String> {
    private final Counter emptyLines = Metrics.counter(ExtractWordsFn.class, "emptyLines");

    @ProcessElement
    public void processElement(ProcessContext c) {
      if (c.element().trim().isEmpty()) {
        emptyLines.inc();
      }

      // Split the line into words.
      String[] words = c.element().split("[^a-zA-Z']+");

      // Output each word encountered into the output PCollection.
      for (String word : words) {
        if (!word.isEmpty()) {
          c.output(word);
        }
      }
    }
  }

  /** A SimpleFunction that converts a Word and Count into a printable string. */
  public static class FormatAsTextFn extends SimpleFunction<KV<String, Long>, String> {
    @Override
    public String apply(KV<String, Long> input) {
      return input.getKey() + ": " + input.getValue();
    }
  }

  /**
   * A PTransform that converts a PCollection containing lines of text into a PCollection of
   * formatted word counts.
   */
  public static class CountWords
      extends PTransform<PCollection<String>, PCollection<KV<String, Long>>> {
    @Override
    public PCollection<KV<String, Long>> expand(PCollection<String> lines) {

      // Convert lines of text into individual words.
      PCollection<String> words = lines.apply(ParDo.of(new ExtractWordsFn()));

      // Count the number of times each word occurs.
      PCollection<KV<String, Long>> wordCounts = words.apply(Count.<String>perElement());

      return wordCounts;
    }
  }

  /**
   * Options supported by {@link com.google.cloud.teleport.templates.WordCount}.
   *
   * <p>Inherits standard configuration options.
   */
  public interface WordCountOptions extends PipelineOptions {
    @Description("Path of the file to read from")
    ValueProvider<String> getInputFile();

    void setInputFile(ValueProvider<String> value);

    @Description("Path of the file to write to")
    ValueProvider<String> getOutput();

    void setOutput(ValueProvider<String> value);
  }

  public static void main(String[] args) {
    WordCountOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(WordCountOptions.class);
    Pipeline p = Pipeline.create(options);
    p.apply("ReadLines", TextIO.read().from(options.getInputFile()))
        .apply(new CountWords())
        .apply(MapElements.via(new FormatAsTextFn()))
        .apply("WriteCounts", TextIO.write().to(options.getOutput()));

    p.run();
  }
}

Extend templates

You can build your own pipelines by extending the open source Dataflow templates. For example, for a template that uses a fixed window duration, data that arrives outside of the window might be discarded. To avoid this behavior, use the template code as a base, and modify the code to invoke the .withAllowedLateness operation.