> expectedResults = Arrays.asList(
+ KV.of("Flourish", 3L),
+ KV.of("stomach", 1L));
+ DataflowAssert.that(filteredWords).containsInAnyOrder(expectedResults);
+
+ p.run();
+ }
+}
diff --git a/examples/src/main/java/com/google/cloud/dataflow/examples/MinimalWordCount.java b/examples/src/main/java/com/google/cloud/dataflow/examples/MinimalWordCount.java
new file mode 100644
index 000000000000..4ed05207c461
--- /dev/null
+++ b/examples/src/main/java/com/google/cloud/dataflow/examples/MinimalWordCount.java
@@ -0,0 +1,117 @@
+/*
+ * Copyright (C) 2015 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.dataflow.examples;
+
+import com.google.cloud.dataflow.sdk.Pipeline;
+import com.google.cloud.dataflow.sdk.io.TextIO;
+import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
+import com.google.cloud.dataflow.sdk.runners.BlockingDataflowPipelineRunner;
+import com.google.cloud.dataflow.sdk.transforms.Count;
+import com.google.cloud.dataflow.sdk.transforms.DoFn;
+import com.google.cloud.dataflow.sdk.transforms.MapElements;
+import com.google.cloud.dataflow.sdk.transforms.ParDo;
+import com.google.cloud.dataflow.sdk.transforms.SimpleFunction;
+import com.google.cloud.dataflow.sdk.values.KV;
+
+
+/**
+ * An example that counts words in Shakespeare.
+ *
+ * This class, {@link MinimalWordCount}, is the first in a series of four successively more
+ * detailed 'word count' examples. Here, for simplicity, we don't show any error-checking or
+ * argument processing, and focus on construction of the pipeline, which chains together the
+ * application of core transforms.
+ *
+ *
Next, see the {@link WordCount} pipeline, then the {@link DebuggingWordCount}, and finally
+ * the {@link WindowedWordCount} pipeline, for more detailed examples that introduce additional
+ * concepts.
+ *
+ *
Concepts:
+ *
+ * 1. Reading data from text files
+ * 2. Specifying 'inline' transforms
+ * 3. Counting a PCollection
+ * 4. Writing data to Cloud Storage as text files
+ *
+ *
+ * To execute this pipeline, first edit the code to set your project ID, the staging
+ * location, and the output location. The specified GCS bucket(s) must already exist.
+ *
+ *
Then, run the pipeline as described in the README. It will be deployed and run using the
+ * Dataflow service. No args are required to run the pipeline. You can see the results in your
+ * output bucket in the GCS browser.
+ */
+public class MinimalWordCount {
+
+ public static void main(String[] args) {
+ // Create a DataflowPipelineOptions object. This object lets us set various execution
+ // options for our pipeline, such as the associated Cloud Platform project and the location
+ // in Google Cloud Storage to stage files.
+ DataflowPipelineOptions options = PipelineOptionsFactory.create()
+ .as(DataflowPipelineOptions.class);
+ options.setRunner(BlockingDataflowPipelineRunner.class);
+ // CHANGE 1/3: Your project ID is required in order to run your pipeline on the Google Cloud.
+ options.setProject("SET_YOUR_PROJECT_ID_HERE");
+ // CHANGE 2/3: Your Google Cloud Storage path is required for staging local files.
+ options.setStagingLocation("gs://SET_YOUR_BUCKET_NAME_HERE/AND_STAGING_DIRECTORY");
+
+ // Create the Pipeline object with the options we defined above.
+ Pipeline p = Pipeline.create(options);
+
+ // Apply the pipeline's transforms.
+
+ // Concept #1: Apply a root transform to the pipeline; in this case, TextIO.Read to read a set
+ // of input text files. TextIO.Read returns a PCollection where each element is one line from
+ // the input text (a set of Shakespeare's texts).
+ p.apply(TextIO.Read.from("gs://dataflow-samples/shakespeare/*"))
+ // Concept #2: Apply a ParDo transform to our PCollection of text lines. This ParDo invokes a
+ // DoFn (defined in-line) on each element that tokenizes the text line into individual words.
+ // The ParDo returns a PCollection, where each element is an individual word in
+ // Shakespeare's collected texts.
+ .apply(ParDo.named("ExtractWords").of(new DoFn() {
+ @Override
+ public void processElement(ProcessContext c) {
+ for (String word : c.element().split("[^a-zA-Z']+")) {
+ if (!word.isEmpty()) {
+ c.output(word);
+ }
+ }
+ }
+ }))
+ // Concept #3: Apply the Count transform to our PCollection of individual words. The Count
+ // transform returns a new PCollection of key/value pairs, where each key represents a unique
+ // word in the text. The associated value is the occurrence count for that word.
+ .apply(Count.perElement())
+ // Apply a MapElements transform that formats our PCollection of word counts into a printable
+ // string, suitable for writing to an output file.
+ .apply("FormatResults", MapElements.via(new SimpleFunction, String>() {
+ @Override
+ public String apply(KV input) {
+ return input.getKey() + ": " + input.getValue();
+ }
+ }))
+ // Concept #4: Apply a write transform, TextIO.Write, at the end of the pipeline.
+ // TextIO.Write writes the contents of a PCollection (in this case, our PCollection of
+ // formatted strings) to a series of text files in Google Cloud Storage.
+ // CHANGE 3/3: The Google Cloud Storage path is required for outputting the results to.
+ .apply(TextIO.Write.to("gs://YOUR_OUTPUT_BUCKET/AND_OUTPUT_PREFIX"));
+
+ // Run the pipeline.
+ p.run();
+ }
+}
diff --git a/examples/src/main/java/com/google/cloud/dataflow/examples/WindowedWordCount.java b/examples/src/main/java/com/google/cloud/dataflow/examples/WindowedWordCount.java
new file mode 100644
index 000000000000..2adac5562731
--- /dev/null
+++ b/examples/src/main/java/com/google/cloud/dataflow/examples/WindowedWordCount.java
@@ -0,0 +1,269 @@
+/*
+ * Copyright (C) 2015 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.dataflow.examples;
+
+import com.google.api.services.bigquery.model.TableFieldSchema;
+import com.google.api.services.bigquery.model.TableReference;
+import com.google.api.services.bigquery.model.TableRow;
+import com.google.api.services.bigquery.model.TableSchema;
+import com.google.cloud.dataflow.examples.common.DataflowExampleOptions;
+import com.google.cloud.dataflow.examples.common.DataflowExampleUtils;
+import com.google.cloud.dataflow.examples.common.ExampleBigQueryTableOptions;
+import com.google.cloud.dataflow.examples.common.ExamplePubsubTopicOptions;
+import com.google.cloud.dataflow.sdk.Pipeline;
+import com.google.cloud.dataflow.sdk.PipelineResult;
+import com.google.cloud.dataflow.sdk.io.BigQueryIO;
+import com.google.cloud.dataflow.sdk.io.PubsubIO;
+import com.google.cloud.dataflow.sdk.io.TextIO;
+import com.google.cloud.dataflow.sdk.options.Default;
+import com.google.cloud.dataflow.sdk.options.Description;
+import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
+import com.google.cloud.dataflow.sdk.transforms.DoFn;
+import com.google.cloud.dataflow.sdk.transforms.ParDo;
+import com.google.cloud.dataflow.sdk.transforms.windowing.FixedWindows;
+import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
+import com.google.cloud.dataflow.sdk.values.KV;
+import com.google.cloud.dataflow.sdk.values.PCollection;
+
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+
+/**
+ * An example that counts words in text, and can run over either unbounded or bounded input
+ * collections.
+ *
+ * This class, {@link WindowedWordCount}, is the last in a series of four successively more
+ * detailed 'word count' examples. First take a look at {@link MinimalWordCount},
+ * {@link WordCount}, and {@link DebuggingWordCount}.
+ *
+ *
Basic concepts, also in the MinimalWordCount, WordCount, and DebuggingWordCount examples:
+ * Reading text files; counting a PCollection; writing to GCS; executing a Pipeline both locally
+ * and using the Dataflow service; defining DoFns; creating a custom aggregator;
+ * user-defined PTransforms; defining PipelineOptions.
+ *
+ *
New Concepts:
+ *
+ * 1. Unbounded and bounded pipeline input modes
+ * 2. Adding timestamps to data
+ * 3. PubSub topics as sources
+ * 4. Windowing
+ * 5. Re-using PTransforms over windowed PCollections
+ * 6. Writing to BigQuery
+ *
+ *
+ * To execute this pipeline locally, specify general pipeline configuration:
+ *
{@code
+ * --project=YOUR_PROJECT_ID
+ * }
+ *
+ *
+ * To execute this pipeline using the Dataflow service, specify pipeline configuration:
+ *
{@code
+ * --project=YOUR_PROJECT_ID
+ * --stagingLocation=gs://YOUR_STAGING_DIRECTORY
+ * --runner=BlockingDataflowPipelineRunner
+ * }
+ *
+ *
+ * Optionally specify the input file path via:
+ * {@code --inputFile=gs://INPUT_PATH},
+ * which defaults to {@code gs://dataflow-samples/shakespeare/kinglear.txt}.
+ *
+ *
Specify an output BigQuery dataset and optionally, a table for the output. If you don't
+ * specify the table, one will be created for you using the job name. If you don't specify the
+ * dataset, a dataset called {@code dataflow-examples} must already exist in your project.
+ * {@code --bigQueryDataset=YOUR-DATASET --bigQueryTable=YOUR-NEW-TABLE-NAME}.
+ *
+ *
Decide whether you want your pipeline to run with 'bounded' (such as files in GCS) or
+ * 'unbounded' input (such as a PubSub topic). To run with unbounded input, set
+ * {@code --unbounded=true}. Then, optionally specify the Google Cloud PubSub topic to read from
+ * via {@code --pubsubTopic=projects/PROJECT_ID/topics/YOUR_TOPIC_NAME}. If the topic does not
+ * exist, the pipeline will create one for you. It will delete this topic when it terminates.
+ * The pipeline will automatically launch an auxiliary batch pipeline to populate the given PubSub
+ * topic with the contents of the {@code --inputFile}, in order to make the example easy to run.
+ * If you want to use an independently-populated PubSub topic, indicate this by setting
+ * {@code --inputFile=""}. In that case, the auxiliary pipeline will not be started.
+ *
+ *
By default, the pipeline will do fixed windowing, on 1-minute windows. You can
+ * change this interval by setting the {@code --windowSize} parameter, e.g. {@code --windowSize=10}
+ * for 10-minute windows.
+ */
+public class WindowedWordCount {
+ private static final Logger LOG = LoggerFactory.getLogger(WindowedWordCount.class);
+ static final int WINDOW_SIZE = 1; // Default window duration in minutes
+
+ /**
+ * Concept #2: A DoFn that sets the data element timestamp. This is a silly method, just for
+ * this example, for the bounded data case.
+ *
+ *
Imagine that many ghosts of Shakespeare are all typing madly at the same time to recreate
+ * his masterworks. Each line of the corpus will get a random associated timestamp somewhere in a
+ * 2-hour period.
+ */
+ static class AddTimestampFn extends DoFn {
+ private static final long RAND_RANGE = 7200000; // 2 hours in ms
+
+ @Override
+ public void processElement(ProcessContext c) {
+ // Generate a timestamp that falls somewhere in the past two hours.
+ long randomTimestamp = System.currentTimeMillis()
+ - (int) (Math.random() * RAND_RANGE);
+ /**
+ * Concept #2: Set the data element with that timestamp.
+ */
+ c.outputWithTimestamp(c.element(), new Instant(randomTimestamp));
+ }
+ }
+
+ /** A DoFn that converts a Word and Count into a BigQuery table row. */
+ static class FormatAsTableRowFn extends DoFn, TableRow> {
+ @Override
+ public void processElement(ProcessContext c) {
+ TableRow row = new TableRow()
+ .set("word", c.element().getKey())
+ .set("count", c.element().getValue())
+ // include a field for the window timestamp
+ .set("window_timestamp", c.timestamp().toString());
+ c.output(row);
+ }
+ }
+
+ /**
+ * Helper method that defines the BigQuery schema used for the output.
+ */
+ private static TableSchema getSchema() {
+ List fields = new ArrayList<>();
+ fields.add(new TableFieldSchema().setName("word").setType("STRING"));
+ fields.add(new TableFieldSchema().setName("count").setType("INTEGER"));
+ fields.add(new TableFieldSchema().setName("window_timestamp").setType("TIMESTAMP"));
+ TableSchema schema = new TableSchema().setFields(fields);
+ return schema;
+ }
+
+ /**
+ * Concept #6: We'll stream the results to a BigQuery table. The BigQuery output source is one
+ * that supports both bounded and unbounded data. This is a helper method that creates a
+ * TableReference from input options, to tell the pipeline where to write its BigQuery results.
+ */
+ private static TableReference getTableReference(Options options) {
+ TableReference tableRef = new TableReference();
+ tableRef.setProjectId(options.getProject());
+ tableRef.setDatasetId(options.getBigQueryDataset());
+ tableRef.setTableId(options.getBigQueryTable());
+ return tableRef;
+ }
+
+ /**
+ * Options supported by {@link WindowedWordCount}.
+ *
+ * Inherits standard example configuration options, which allow specification of the BigQuery
+ * table and the PubSub topic, as well as the {@link WordCount.WordCountOptions} support for
+ * specification of the input file.
+ */
+ public static interface Options extends WordCount.WordCountOptions,
+ DataflowExampleOptions, ExamplePubsubTopicOptions, ExampleBigQueryTableOptions {
+ @Description("Fixed window duration, in minutes")
+ @Default.Integer(WINDOW_SIZE)
+ Integer getWindowSize();
+ void setWindowSize(Integer value);
+
+ @Description("Whether to run the pipeline with unbounded input")
+ boolean isUnbounded();
+ void setUnbounded(boolean value);
+ }
+
+ public static void main(String[] args) throws IOException {
+ Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
+ options.setBigQuerySchema(getSchema());
+ // DataflowExampleUtils creates the necessary input sources to simplify execution of this
+ // Pipeline.
+ DataflowExampleUtils exampleDataflowUtils = new DataflowExampleUtils(options,
+ options.isUnbounded());
+
+ Pipeline pipeline = Pipeline.create(options);
+
+ /**
+ * Concept #1: the Dataflow SDK lets us run the same pipeline with either a bounded or
+ * unbounded input source.
+ */
+ PCollection input;
+ if (options.isUnbounded()) {
+ LOG.info("Reading from PubSub.");
+ /**
+ * Concept #3: Read from the PubSub topic. A topic will be created if it wasn't
+ * specified as an argument. The data elements' timestamps will come from the pubsub
+ * injection.
+ */
+ input = pipeline
+ .apply(PubsubIO.Read.topic(options.getPubsubTopic()));
+ } else {
+ /** Else, this is a bounded pipeline. Read from the GCS file. */
+ input = pipeline
+ .apply(TextIO.Read.from(options.getInputFile()))
+ // Concept #2: Add an element timestamp, using an artificial time just to show windowing.
+ // See AddTimestampFn for more detail on this.
+ .apply(ParDo.of(new AddTimestampFn()));
+ }
+
+ /**
+ * Concept #4: Window into fixed windows. The fixed window size for this example defaults to 1
+ * minute (you can change this with a command-line option). See the documentation for more
+ * information on how fixed windows work, and for information on the other types of windowing
+ * available (e.g., sliding windows).
+ */
+ PCollection windowedWords = input
+ .apply(Window.into(
+ FixedWindows.of(Duration.standardMinutes(options.getWindowSize()))));
+
+ /**
+ * Concept #5: Re-use our existing CountWords transform that does not have knowledge of
+ * windows over a PCollection containing windowed values.
+ */
+ PCollection> wordCounts = windowedWords.apply(new WordCount.CountWords());
+
+ /**
+ * Concept #6: Format the results for a BigQuery table, then write to BigQuery.
+ * The BigQuery output source supports both bounded and unbounded data.
+ */
+ wordCounts.apply(ParDo.of(new FormatAsTableRowFn()))
+ .apply(BigQueryIO.Write
+ .to(getTableReference(options))
+ .withSchema(getSchema())
+ .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
+ .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));
+
+ PipelineResult result = pipeline.run();
+
+ /**
+ * To mock unbounded input from PubSub, we'll now start an auxiliary 'injector' pipeline that
+ * runs for a limited time, and publishes to the input PubSub topic.
+ *
+ * With an unbounded input source, you will need to explicitly shut down this pipeline when you
+ * are done with it, so that you do not continue to be charged for the instances. You can do
+ * this via a ctrl-C from the command line, or from the developer's console UI for Dataflow
+ * pipelines. The PubSub topic will also be deleted at this time.
+ */
+ exampleDataflowUtils.mockUnboundedSource(options.getInputFile(), result);
+ }
+}
diff --git a/examples/src/main/java/com/google/cloud/dataflow/examples/WordCount.java b/examples/src/main/java/com/google/cloud/dataflow/examples/WordCount.java
new file mode 100644
index 000000000000..1086106f0498
--- /dev/null
+++ b/examples/src/main/java/com/google/cloud/dataflow/examples/WordCount.java
@@ -0,0 +1,206 @@
+/*
+ * Copyright (C) 2015 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.dataflow.examples;
+
+import com.google.cloud.dataflow.sdk.Pipeline;
+import com.google.cloud.dataflow.sdk.io.TextIO;
+import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.options.Default;
+import com.google.cloud.dataflow.sdk.options.DefaultValueFactory;
+import com.google.cloud.dataflow.sdk.options.Description;
+import com.google.cloud.dataflow.sdk.options.PipelineOptions;
+import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
+import com.google.cloud.dataflow.sdk.transforms.Aggregator;
+import com.google.cloud.dataflow.sdk.transforms.Count;
+import com.google.cloud.dataflow.sdk.transforms.DoFn;
+import com.google.cloud.dataflow.sdk.transforms.MapElements;
+import com.google.cloud.dataflow.sdk.transforms.PTransform;
+import com.google.cloud.dataflow.sdk.transforms.ParDo;
+import com.google.cloud.dataflow.sdk.transforms.SimpleFunction;
+import com.google.cloud.dataflow.sdk.transforms.Sum;
+import com.google.cloud.dataflow.sdk.util.gcsfs.GcsPath;
+import com.google.cloud.dataflow.sdk.values.KV;
+import com.google.cloud.dataflow.sdk.values.PCollection;
+
+
+/**
+ * An example that counts words in Shakespeare and includes Dataflow best practices.
+ *
+ * This class, {@link WordCount}, is the second in a series of four successively more detailed
+ * 'word count' examples. You may first want to take a look at {@link MinimalWordCount}.
+ * After you've looked at this example, then see the {@link DebuggingWordCount}
+ * pipeline, for introduction of additional concepts.
+ *
+ *
For a detailed walkthrough of this example, see
+ *
+ * https://cloud.google.com/dataflow/java-sdk/wordcount-example
+ *
+ *
+ *
Basic concepts, also in the MinimalWordCount example:
+ * Reading text files; counting a PCollection; writing to GCS.
+ *
+ *
New Concepts:
+ *
+ * 1. Executing a Pipeline both locally and using the Dataflow service
+ * 2. Using ParDo with static DoFns defined out-of-line
+ * 3. Building a composite transform
+ * 4. Defining your own pipeline options
+ *
+ *
+ * Concept #1: you can execute this pipeline either locally or using the Dataflow service.
+ * These are now command-line options and not hard-coded as they were in the MinimalWordCount
+ * example.
+ * To execute this pipeline locally, specify general pipeline configuration:
+ *
{@code
+ * --project=YOUR_PROJECT_ID
+ * }
+ *
+ * and a local output file or output prefix on GCS:
+ * {@code
+ * --output=[YOUR_LOCAL_FILE | gs://YOUR_OUTPUT_PREFIX]
+ * }
+ *
+ * To execute this pipeline using the Dataflow service, specify pipeline configuration:
+ *
{@code
+ * --project=YOUR_PROJECT_ID
+ * --stagingLocation=gs://YOUR_STAGING_DIRECTORY
+ * --runner=BlockingDataflowPipelineRunner
+ * }
+ *
+ * and an output prefix on GCS:
+ * {@code
+ * --output=gs://YOUR_OUTPUT_PREFIX
+ * }
+ *
+ * The input file defaults to {@code gs://dataflow-samples/shakespeare/kinglear.txt} and can be
+ * overridden with {@code --inputFile}.
+ */
+public class WordCount {
+
+ /**
+ * Concept #2: You can make your pipeline code less verbose by defining your DoFns statically out-
+ * of-line. This DoFn tokenizes lines of text into individual words; we pass it to a ParDo in the
+ * pipeline.
+ */
+ static class ExtractWordsFn extends DoFn {
+ private final Aggregator emptyLines =
+ createAggregator("emptyLines", new Sum.SumLongFn());
+
+ @Override
+ public void processElement(ProcessContext c) {
+ if (c.element().trim().isEmpty()) {
+ emptyLines.addValue(1L);
+ }
+
+ // 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, String> {
+ @Override
+ public String apply(KV input) {
+ return input.getKey() + ": " + input.getValue();
+ }
+ }
+
+ /**
+ * A PTransform that converts a PCollection containing lines of text into a PCollection of
+ * formatted word counts.
+ *
+ * Concept #3: This is a custom composite transform that bundles two transforms (ParDo and
+ * Count) as a reusable PTransform subclass. Using composite transforms allows for easy reuse,
+ * modular testing, and an improved monitoring experience.
+ */
+ public static class CountWords extends PTransform,
+ PCollection>> {
+ @Override
+ public PCollection> apply(PCollection lines) {
+
+ // Convert lines of text into individual words.
+ PCollection words = lines.apply(
+ ParDo.of(new ExtractWordsFn()));
+
+ // Count the number of times each word occurs.
+ PCollection> wordCounts =
+ words.apply(Count.perElement());
+
+ return wordCounts;
+ }
+ }
+
+ /**
+ * Options supported by {@link WordCount}.
+ *
+ * Concept #4: Defining your own configuration options. Here, you can add your own arguments
+ * to be processed by the command-line parser, and specify default values for them. You can then
+ * access the options values in your pipeline code.
+ *
+ *
Inherits standard configuration options.
+ */
+ public static interface WordCountOptions extends PipelineOptions {
+ @Description("Path of the file to read from")
+ @Default.String("gs://dataflow-samples/shakespeare/kinglear.txt")
+ String getInputFile();
+ void setInputFile(String value);
+
+ @Description("Path of the file to write to")
+ @Default.InstanceFactory(OutputFactory.class)
+ String getOutput();
+ void setOutput(String value);
+
+ /**
+ * Returns "gs://${YOUR_STAGING_DIRECTORY}/counts.txt" as the default destination.
+ */
+ public static class OutputFactory implements DefaultValueFactory {
+ @Override
+ public String create(PipelineOptions options) {
+ DataflowPipelineOptions dataflowOptions = options.as(DataflowPipelineOptions.class);
+ if (dataflowOptions.getStagingLocation() != null) {
+ return GcsPath.fromUri(dataflowOptions.getStagingLocation())
+ .resolve("counts.txt").toString();
+ } else {
+ throw new IllegalArgumentException("Must specify --output or --stagingLocation");
+ }
+ }
+ }
+
+ }
+
+ public static void main(String[] args) {
+ WordCountOptions options = PipelineOptionsFactory.fromArgs(args).withValidation()
+ .as(WordCountOptions.class);
+ Pipeline p = Pipeline.create(options);
+
+ // Concepts #2 and #3: Our pipeline applies the composite CountWords transform, and passes the
+ // static FormatAsTextFn() to the ParDo transform.
+ p.apply(TextIO.Read.named("ReadLines").from(options.getInputFile()))
+ .apply(new CountWords())
+ .apply(MapElements.via(new FormatAsTextFn()))
+ .apply(TextIO.Write.named("WriteCounts").to(options.getOutput()));
+
+ p.run();
+ }
+}
diff --git a/examples/src/main/java/com/google/cloud/dataflow/examples/common/DataflowExampleOptions.java b/examples/src/main/java/com/google/cloud/dataflow/examples/common/DataflowExampleOptions.java
new file mode 100644
index 000000000000..606bfb4c03e9
--- /dev/null
+++ b/examples/src/main/java/com/google/cloud/dataflow/examples/common/DataflowExampleOptions.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright (C) 2015 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.dataflow.examples.common;
+
+import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.options.Default;
+import com.google.cloud.dataflow.sdk.options.Description;
+
+/**
+ * Options that can be used to configure the Dataflow examples.
+ */
+public interface DataflowExampleOptions extends DataflowPipelineOptions {
+ @Description("Whether to keep jobs running on the Dataflow service after local process exit")
+ @Default.Boolean(false)
+ boolean getKeepJobsRunning();
+ void setKeepJobsRunning(boolean keepJobsRunning);
+
+ @Description("Number of workers to use when executing the injector pipeline")
+ @Default.Integer(1)
+ int getInjectorNumWorkers();
+ void setInjectorNumWorkers(int numWorkers);
+}
diff --git a/examples/src/main/java/com/google/cloud/dataflow/examples/common/DataflowExampleUtils.java b/examples/src/main/java/com/google/cloud/dataflow/examples/common/DataflowExampleUtils.java
new file mode 100644
index 000000000000..4dfdd85b803a
--- /dev/null
+++ b/examples/src/main/java/com/google/cloud/dataflow/examples/common/DataflowExampleUtils.java
@@ -0,0 +1,485 @@
+/*
+ * Copyright (C) 2015 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.dataflow.examples.common;
+
+import com.google.api.client.googleapis.json.GoogleJsonResponseException;
+import com.google.api.client.googleapis.services.AbstractGoogleClientRequest;
+import com.google.api.client.util.BackOff;
+import com.google.api.client.util.BackOffUtils;
+import com.google.api.client.util.Sleeper;
+import com.google.api.services.bigquery.Bigquery;
+import com.google.api.services.bigquery.Bigquery.Datasets;
+import com.google.api.services.bigquery.Bigquery.Tables;
+import com.google.api.services.bigquery.model.Dataset;
+import com.google.api.services.bigquery.model.DatasetReference;
+import com.google.api.services.bigquery.model.Table;
+import com.google.api.services.bigquery.model.TableReference;
+import com.google.api.services.bigquery.model.TableSchema;
+import com.google.api.services.dataflow.Dataflow;
+import com.google.api.services.pubsub.Pubsub;
+import com.google.api.services.pubsub.model.Subscription;
+import com.google.api.services.pubsub.model.Topic;
+import com.google.cloud.dataflow.sdk.Pipeline;
+import com.google.cloud.dataflow.sdk.PipelineResult;
+import com.google.cloud.dataflow.sdk.io.TextIO;
+import com.google.cloud.dataflow.sdk.options.BigQueryOptions;
+import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.runners.DataflowPipelineJob;
+import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
+import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
+import com.google.cloud.dataflow.sdk.transforms.IntraBundleParallelization;
+import com.google.cloud.dataflow.sdk.transforms.PTransform;
+import com.google.cloud.dataflow.sdk.util.AttemptBoundedExponentialBackOff;
+import com.google.cloud.dataflow.sdk.util.MonitoringUtil;
+import com.google.cloud.dataflow.sdk.util.Transport;
+import com.google.cloud.dataflow.sdk.values.PBegin;
+import com.google.cloud.dataflow.sdk.values.PCollection;
+import com.google.common.base.Strings;
+import com.google.common.base.Throwables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import javax.servlet.http.HttpServletResponse;
+
+/**
+ * The utility class that sets up and tears down external resources, starts the Google Cloud Pub/Sub
+ * injector, and cancels the streaming and the injector pipelines once the program terminates.
+ *
+ * It is used to run Dataflow examples, such as TrafficMaxLaneFlow and TrafficRoutes.
+ */
+public class DataflowExampleUtils {
+
+ private final DataflowPipelineOptions options;
+ private Bigquery bigQueryClient = null;
+ private Pubsub pubsubClient = null;
+ private Dataflow dataflowClient = null;
+ private Set jobsToCancel = Sets.newHashSet();
+ private List pendingMessages = Lists.newArrayList();
+
+ public DataflowExampleUtils(DataflowPipelineOptions options) {
+ this.options = options;
+ }
+
+ /**
+ * Do resources and runner options setup.
+ */
+ public DataflowExampleUtils(DataflowPipelineOptions options, boolean isUnbounded)
+ throws IOException {
+ this.options = options;
+ setupResourcesAndRunner(isUnbounded);
+ }
+
+ /**
+ * Sets up external resources that are required by the example,
+ * such as Pub/Sub topics and BigQuery tables.
+ *
+ * @throws IOException if there is a problem setting up the resources
+ */
+ public void setup() throws IOException {
+ Sleeper sleeper = Sleeper.DEFAULT;
+ BackOff backOff = new AttemptBoundedExponentialBackOff(3, 200);
+ Throwable lastException = null;
+ try {
+ do {
+ try {
+ setupPubsub();
+ setupBigQueryTable();
+ return;
+ } catch (GoogleJsonResponseException e) {
+ lastException = e;
+ }
+ } while (BackOffUtils.next(sleeper, backOff));
+ } catch (InterruptedException e) {
+ // Ignore InterruptedException
+ }
+ Throwables.propagate(lastException);
+ }
+
+ /**
+ * Set up external resources, and configure the runner appropriately.
+ */
+ public void setupResourcesAndRunner(boolean isUnbounded) throws IOException {
+ if (isUnbounded) {
+ options.setStreaming(true);
+ }
+ setup();
+ setupRunner();
+ }
+
+ /**
+ * Sets up the Google Cloud Pub/Sub topic.
+ *
+ * If the topic doesn't exist, a new topic with the given name will be created.
+ *
+ * @throws IOException if there is a problem setting up the Pub/Sub topic
+ */
+ public void setupPubsub() throws IOException {
+ ExamplePubsubTopicAndSubscriptionOptions pubsubOptions =
+ options.as(ExamplePubsubTopicAndSubscriptionOptions.class);
+ if (!pubsubOptions.getPubsubTopic().isEmpty()) {
+ pendingMessages.add("**********************Set Up Pubsub************************");
+ setupPubsubTopic(pubsubOptions.getPubsubTopic());
+ pendingMessages.add("The Pub/Sub topic has been set up for this example: "
+ + pubsubOptions.getPubsubTopic());
+
+ if (!pubsubOptions.getPubsubSubscription().isEmpty()) {
+ setupPubsubSubscription(
+ pubsubOptions.getPubsubTopic(), pubsubOptions.getPubsubSubscription());
+ pendingMessages.add("The Pub/Sub subscription has been set up for this example: "
+ + pubsubOptions.getPubsubSubscription());
+ }
+ }
+ }
+
+ /**
+ * Sets up the BigQuery table with the given schema.
+ *
+ *
If the table already exists, the schema has to match the given one. Otherwise, the example
+ * will throw a RuntimeException. If the table doesn't exist, a new table with the given schema
+ * will be created.
+ *
+ * @throws IOException if there is a problem setting up the BigQuery table
+ */
+ public void setupBigQueryTable() throws IOException {
+ ExampleBigQueryTableOptions bigQueryTableOptions =
+ options.as(ExampleBigQueryTableOptions.class);
+ if (bigQueryTableOptions.getBigQueryDataset() != null
+ && bigQueryTableOptions.getBigQueryTable() != null
+ && bigQueryTableOptions.getBigQuerySchema() != null) {
+ pendingMessages.add("******************Set Up Big Query Table*******************");
+ setupBigQueryTable(bigQueryTableOptions.getProject(),
+ bigQueryTableOptions.getBigQueryDataset(),
+ bigQueryTableOptions.getBigQueryTable(),
+ bigQueryTableOptions.getBigQuerySchema());
+ pendingMessages.add("The BigQuery table has been set up for this example: "
+ + bigQueryTableOptions.getProject()
+ + ":" + bigQueryTableOptions.getBigQueryDataset()
+ + "." + bigQueryTableOptions.getBigQueryTable());
+ }
+ }
+
+ /**
+ * Tears down external resources that can be deleted upon the example's completion.
+ */
+ private void tearDown() {
+ pendingMessages.add("*************************Tear Down*************************");
+ ExamplePubsubTopicAndSubscriptionOptions pubsubOptions =
+ options.as(ExamplePubsubTopicAndSubscriptionOptions.class);
+ if (!pubsubOptions.getPubsubTopic().isEmpty()) {
+ try {
+ deletePubsubTopic(pubsubOptions.getPubsubTopic());
+ pendingMessages.add("The Pub/Sub topic has been deleted: "
+ + pubsubOptions.getPubsubTopic());
+ } catch (IOException e) {
+ pendingMessages.add("Failed to delete the Pub/Sub topic : "
+ + pubsubOptions.getPubsubTopic());
+ }
+ if (!pubsubOptions.getPubsubSubscription().isEmpty()) {
+ try {
+ deletePubsubSubscription(pubsubOptions.getPubsubSubscription());
+ pendingMessages.add("The Pub/Sub subscription has been deleted: "
+ + pubsubOptions.getPubsubSubscription());
+ } catch (IOException e) {
+ pendingMessages.add("Failed to delete the Pub/Sub subscription : "
+ + pubsubOptions.getPubsubSubscription());
+ }
+ }
+ }
+
+ ExampleBigQueryTableOptions bigQueryTableOptions =
+ options.as(ExampleBigQueryTableOptions.class);
+ if (bigQueryTableOptions.getBigQueryDataset() != null
+ && bigQueryTableOptions.getBigQueryTable() != null
+ && bigQueryTableOptions.getBigQuerySchema() != null) {
+ pendingMessages.add("The BigQuery table might contain the example's output, "
+ + "and it is not deleted automatically: "
+ + bigQueryTableOptions.getProject()
+ + ":" + bigQueryTableOptions.getBigQueryDataset()
+ + "." + bigQueryTableOptions.getBigQueryTable());
+ pendingMessages.add("Please go to the Developers Console to delete it manually."
+ + " Otherwise, you may be charged for its usage.");
+ }
+ }
+
+ private void setupBigQueryTable(String projectId, String datasetId, String tableId,
+ TableSchema schema) throws IOException {
+ if (bigQueryClient == null) {
+ bigQueryClient = Transport.newBigQueryClient(options.as(BigQueryOptions.class)).build();
+ }
+
+ Datasets datasetService = bigQueryClient.datasets();
+ if (executeNullIfNotFound(datasetService.get(projectId, datasetId)) == null) {
+ Dataset newDataset = new Dataset().setDatasetReference(
+ new DatasetReference().setProjectId(projectId).setDatasetId(datasetId));
+ datasetService.insert(projectId, newDataset).execute();
+ }
+
+ Tables tableService = bigQueryClient.tables();
+ Table table = executeNullIfNotFound(tableService.get(projectId, datasetId, tableId));
+ if (table == null) {
+ Table newTable = new Table().setSchema(schema).setTableReference(
+ new TableReference().setProjectId(projectId).setDatasetId(datasetId).setTableId(tableId));
+ tableService.insert(projectId, datasetId, newTable).execute();
+ } else if (!table.getSchema().equals(schema)) {
+ throw new RuntimeException(
+ "Table exists and schemas do not match, expecting: " + schema.toPrettyString()
+ + ", actual: " + table.getSchema().toPrettyString());
+ }
+ }
+
+ private void setupPubsubTopic(String topic) throws IOException {
+ if (pubsubClient == null) {
+ pubsubClient = Transport.newPubsubClient(options).build();
+ }
+ if (executeNullIfNotFound(pubsubClient.projects().topics().get(topic)) == null) {
+ pubsubClient.projects().topics().create(topic, new Topic().setName(topic)).execute();
+ }
+ }
+
+ private void setupPubsubSubscription(String topic, String subscription) throws IOException {
+ if (pubsubClient == null) {
+ pubsubClient = Transport.newPubsubClient(options).build();
+ }
+ if (executeNullIfNotFound(pubsubClient.projects().subscriptions().get(subscription)) == null) {
+ Subscription subInfo = new Subscription()
+ .setAckDeadlineSeconds(60)
+ .setTopic(topic);
+ pubsubClient.projects().subscriptions().create(subscription, subInfo).execute();
+ }
+ }
+
+ /**
+ * Deletes the Google Cloud Pub/Sub topic.
+ *
+ * @throws IOException if there is a problem deleting the Pub/Sub topic
+ */
+ private void deletePubsubTopic(String topic) throws IOException {
+ if (pubsubClient == null) {
+ pubsubClient = Transport.newPubsubClient(options).build();
+ }
+ if (executeNullIfNotFound(pubsubClient.projects().topics().get(topic)) != null) {
+ pubsubClient.projects().topics().delete(topic).execute();
+ }
+ }
+
+ /**
+ * Deletes the Google Cloud Pub/Sub subscription.
+ *
+ * @throws IOException if there is a problem deleting the Pub/Sub subscription
+ */
+ private void deletePubsubSubscription(String subscription) throws IOException {
+ if (pubsubClient == null) {
+ pubsubClient = Transport.newPubsubClient(options).build();
+ }
+ if (executeNullIfNotFound(pubsubClient.projects().subscriptions().get(subscription)) != null) {
+ pubsubClient.projects().subscriptions().delete(subscription).execute();
+ }
+ }
+
+ /**
+ * If this is an unbounded (streaming) pipeline, and both inputFile and pubsub topic are defined,
+ * start an 'injector' pipeline that publishes the contents of the file to the given topic, first
+ * creating the topic if necessary.
+ */
+ public void startInjectorIfNeeded(String inputFile) {
+ ExamplePubsubTopicOptions pubsubTopicOptions = options.as(ExamplePubsubTopicOptions.class);
+ if (pubsubTopicOptions.isStreaming()
+ && !Strings.isNullOrEmpty(inputFile)
+ && !Strings.isNullOrEmpty(pubsubTopicOptions.getPubsubTopic())) {
+ runInjectorPipeline(inputFile, pubsubTopicOptions.getPubsubTopic());
+ }
+ }
+
+ /**
+ * Do some runner setup: check that the DirectPipelineRunner is not used in conjunction with
+ * streaming, and if streaming is specified, use the DataflowPipelineRunner. Return the streaming
+ * flag value.
+ */
+ public void setupRunner() {
+ if (options.isStreaming() && options.getRunner() != DirectPipelineRunner.class) {
+ // In order to cancel the pipelines automatically,
+ // {@literal DataflowPipelineRunner} is forced to be used.
+ options.setRunner(DataflowPipelineRunner.class);
+ }
+ }
+
+ /**
+ * Runs a batch pipeline to inject data into the PubSubIO input topic.
+ *
+ *
The injector pipeline will read from the given text file, and inject data
+ * into the Google Cloud Pub/Sub topic.
+ */
+ public void runInjectorPipeline(String inputFile, String topic) {
+ runInjectorPipeline(TextIO.Read.from(inputFile), topic, null);
+ }
+
+ /**
+ * Runs a batch pipeline to inject data into the PubSubIO input topic.
+ *
+ *
The injector pipeline will read from the given source, and inject data
+ * into the Google Cloud Pub/Sub topic.
+ */
+ public void runInjectorPipeline(PTransform> readSource,
+ String topic,
+ String pubsubTimestampTabelKey) {
+ PubsubFileInjector.Bound injector;
+ if (Strings.isNullOrEmpty(pubsubTimestampTabelKey)) {
+ injector = PubsubFileInjector.publish(topic);
+ } else {
+ injector = PubsubFileInjector.withTimestampLabelKey(pubsubTimestampTabelKey).publish(topic);
+ }
+ DataflowPipelineOptions copiedOptions = options.cloneAs(DataflowPipelineOptions.class);
+ if (options.getServiceAccountName() != null) {
+ copiedOptions.setServiceAccountName(options.getServiceAccountName());
+ }
+ if (options.getServiceAccountKeyfile() != null) {
+ copiedOptions.setServiceAccountKeyfile(options.getServiceAccountKeyfile());
+ }
+ copiedOptions.setStreaming(false);
+ copiedOptions.setNumWorkers(options.as(DataflowExampleOptions.class).getInjectorNumWorkers());
+ copiedOptions.setJobName(options.getJobName() + "-injector");
+ Pipeline injectorPipeline = Pipeline.create(copiedOptions);
+ injectorPipeline.apply(readSource)
+ .apply(IntraBundleParallelization
+ .of(injector)
+ .withMaxParallelism(20));
+ PipelineResult result = injectorPipeline.run();
+ if (result instanceof DataflowPipelineJob) {
+ jobsToCancel.add(((DataflowPipelineJob) result));
+ }
+ }
+
+ /**
+ * Runs the provided pipeline to inject data into the PubSubIO input topic.
+ */
+ public void runInjectorPipeline(Pipeline injectorPipeline) {
+ PipelineResult result = injectorPipeline.run();
+ if (result instanceof DataflowPipelineJob) {
+ jobsToCancel.add(((DataflowPipelineJob) result));
+ }
+ }
+
+ /**
+ * Start the auxiliary injector pipeline, then wait for this pipeline to finish.
+ */
+ public void mockUnboundedSource(String inputFile, PipelineResult result) {
+ startInjectorIfNeeded(inputFile);
+ waitToFinish(result);
+ }
+
+ /**
+ * If {@literal DataflowPipelineRunner} or {@literal BlockingDataflowPipelineRunner} is used,
+ * waits for the pipeline to finish and cancels it (and the injector) before the program exists.
+ */
+ public void waitToFinish(PipelineResult result) {
+ if (result instanceof DataflowPipelineJob) {
+ final DataflowPipelineJob job = (DataflowPipelineJob) result;
+ jobsToCancel.add(job);
+ if (!options.as(DataflowExampleOptions.class).getKeepJobsRunning()) {
+ addShutdownHook(jobsToCancel);
+ }
+ try {
+ job.waitToFinish(-1, TimeUnit.SECONDS, new MonitoringUtil.PrintHandler(System.out));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to wait for job to finish: " + job.getJobId());
+ }
+ } else {
+ // Do nothing if the given PipelineResult doesn't support waitToFinish(),
+ // such as EvaluationResults returned by DirectPipelineRunner.
+ tearDown();
+ printPendingMessages();
+ }
+ }
+
+ private void addShutdownHook(final Collection jobs) {
+ if (dataflowClient == null) {
+ dataflowClient = options.getDataflowClient();
+ }
+
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ tearDown();
+ printPendingMessages();
+ for (DataflowPipelineJob job : jobs) {
+ System.out.println("Canceling example pipeline: " + job.getJobId());
+ try {
+ job.cancel();
+ } catch (IOException e) {
+ System.out.println("Failed to cancel the job,"
+ + " please go to the Developers Console to cancel it manually");
+ System.out.println(
+ MonitoringUtil.getJobMonitoringPageURL(job.getProjectId(), job.getJobId()));
+ }
+ }
+
+ for (DataflowPipelineJob job : jobs) {
+ boolean cancellationVerified = false;
+ for (int retryAttempts = 6; retryAttempts > 0; retryAttempts--) {
+ if (job.getState().isTerminal()) {
+ cancellationVerified = true;
+ System.out.println("Canceled example pipeline: " + job.getJobId());
+ break;
+ } else {
+ System.out.println(
+ "The example pipeline is still running. Verifying the cancellation.");
+ }
+ try {
+ Thread.sleep(10000);
+ } catch (InterruptedException e) {
+ // Ignore
+ }
+ }
+ if (!cancellationVerified) {
+ System.out.println("Failed to verify the cancellation for job: " + job.getJobId());
+ System.out.println("Please go to the Developers Console to verify manually:");
+ System.out.println(
+ MonitoringUtil.getJobMonitoringPageURL(job.getProjectId(), job.getJobId()));
+ }
+ }
+ }
+ });
+ }
+
+ private void printPendingMessages() {
+ System.out.println();
+ System.out.println("***********************************************************");
+ System.out.println("***********************************************************");
+ for (String message : pendingMessages) {
+ System.out.println(message);
+ }
+ System.out.println("***********************************************************");
+ System.out.println("***********************************************************");
+ }
+
+ private static T executeNullIfNotFound(
+ AbstractGoogleClientRequest request) throws IOException {
+ try {
+ return request.execute();
+ } catch (GoogleJsonResponseException e) {
+ if (e.getStatusCode() == HttpServletResponse.SC_NOT_FOUND) {
+ return null;
+ } else {
+ throw e;
+ }
+ }
+ }
+}
diff --git a/examples/src/main/java/com/google/cloud/dataflow/examples/common/ExampleBigQueryTableOptions.java b/examples/src/main/java/com/google/cloud/dataflow/examples/common/ExampleBigQueryTableOptions.java
new file mode 100644
index 000000000000..7c213b59d681
--- /dev/null
+++ b/examples/src/main/java/com/google/cloud/dataflow/examples/common/ExampleBigQueryTableOptions.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright (C) 2015 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.dataflow.examples.common;
+
+import com.google.api.services.bigquery.model.TableSchema;
+import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.options.Default;
+import com.google.cloud.dataflow.sdk.options.DefaultValueFactory;
+import com.google.cloud.dataflow.sdk.options.Description;
+import com.google.cloud.dataflow.sdk.options.PipelineOptions;
+
+/**
+ * Options that can be used to configure BigQuery tables in Dataflow examples.
+ * The project defaults to the project being used to run the example.
+ */
+public interface ExampleBigQueryTableOptions extends DataflowPipelineOptions {
+ @Description("BigQuery dataset name")
+ @Default.String("dataflow_examples")
+ String getBigQueryDataset();
+ void setBigQueryDataset(String dataset);
+
+ @Description("BigQuery table name")
+ @Default.InstanceFactory(BigQueryTableFactory.class)
+ String getBigQueryTable();
+ void setBigQueryTable(String table);
+
+ @Description("BigQuery table schema")
+ TableSchema getBigQuerySchema();
+ void setBigQuerySchema(TableSchema schema);
+
+ /**
+ * Returns the job name as the default BigQuery table name.
+ */
+ static class BigQueryTableFactory implements DefaultValueFactory {
+ @Override
+ public String create(PipelineOptions options) {
+ return options.as(DataflowPipelineOptions.class).getJobName()
+ .replace('-', '_');
+ }
+ }
+}
diff --git a/examples/src/main/java/com/google/cloud/dataflow/examples/common/ExamplePubsubTopicAndSubscriptionOptions.java b/examples/src/main/java/com/google/cloud/dataflow/examples/common/ExamplePubsubTopicAndSubscriptionOptions.java
new file mode 100644
index 000000000000..d7bd4b8edc3d
--- /dev/null
+++ b/examples/src/main/java/com/google/cloud/dataflow/examples/common/ExamplePubsubTopicAndSubscriptionOptions.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright (C) 2015 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.dataflow.examples.common;
+
+import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.options.Default;
+import com.google.cloud.dataflow.sdk.options.DefaultValueFactory;
+import com.google.cloud.dataflow.sdk.options.Description;
+import com.google.cloud.dataflow.sdk.options.PipelineOptions;
+
+/**
+ * Options that can be used to configure Pub/Sub topic/subscription in Dataflow examples.
+ */
+public interface ExamplePubsubTopicAndSubscriptionOptions extends ExamplePubsubTopicOptions {
+ @Description("Pub/Sub subscription")
+ @Default.InstanceFactory(PubsubSubscriptionFactory.class)
+ String getPubsubSubscription();
+ void setPubsubSubscription(String subscription);
+
+ /**
+ * Returns a default Pub/Sub subscription based on the project and the job names.
+ */
+ static class PubsubSubscriptionFactory implements DefaultValueFactory {
+ @Override
+ public String create(PipelineOptions options) {
+ DataflowPipelineOptions dataflowPipelineOptions =
+ options.as(DataflowPipelineOptions.class);
+ return "projects/" + dataflowPipelineOptions.getProject()
+ + "/subscriptions/" + dataflowPipelineOptions.getJobName();
+ }
+ }
+}
diff --git a/examples/src/main/java/com/google/cloud/dataflow/examples/common/ExamplePubsubTopicOptions.java b/examples/src/main/java/com/google/cloud/dataflow/examples/common/ExamplePubsubTopicOptions.java
new file mode 100644
index 000000000000..4bedf318ef5a
--- /dev/null
+++ b/examples/src/main/java/com/google/cloud/dataflow/examples/common/ExamplePubsubTopicOptions.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright (C) 2015 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.dataflow.examples.common;
+
+import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.options.Default;
+import com.google.cloud.dataflow.sdk.options.DefaultValueFactory;
+import com.google.cloud.dataflow.sdk.options.Description;
+import com.google.cloud.dataflow.sdk.options.PipelineOptions;
+
+/**
+ * Options that can be used to configure Pub/Sub topic in Dataflow examples.
+ */
+public interface ExamplePubsubTopicOptions extends DataflowPipelineOptions {
+ @Description("Pub/Sub topic")
+ @Default.InstanceFactory(PubsubTopicFactory.class)
+ String getPubsubTopic();
+ void setPubsubTopic(String topic);
+
+ /**
+ * Returns a default Pub/Sub topic based on the project and the job names.
+ */
+ static class PubsubTopicFactory implements DefaultValueFactory {
+ @Override
+ public String create(PipelineOptions options) {
+ DataflowPipelineOptions dataflowPipelineOptions =
+ options.as(DataflowPipelineOptions.class);
+ return "projects/" + dataflowPipelineOptions.getProject()
+ + "/topics/" + dataflowPipelineOptions.getJobName();
+ }
+ }
+}
diff --git a/examples/src/main/java/com/google/cloud/dataflow/examples/common/PubsubFileInjector.java b/examples/src/main/java/com/google/cloud/dataflow/examples/common/PubsubFileInjector.java
new file mode 100644
index 000000000000..4a82ae612ae7
--- /dev/null
+++ b/examples/src/main/java/com/google/cloud/dataflow/examples/common/PubsubFileInjector.java
@@ -0,0 +1,153 @@
+/*
+ * Copyright (C) 2015 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.dataflow.examples.common;
+
+import com.google.api.services.pubsub.Pubsub;
+import com.google.api.services.pubsub.model.PublishRequest;
+import com.google.api.services.pubsub.model.PubsubMessage;
+import com.google.cloud.dataflow.sdk.Pipeline;
+import com.google.cloud.dataflow.sdk.io.TextIO;
+import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.options.Description;
+import com.google.cloud.dataflow.sdk.options.PipelineOptions;
+import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
+import com.google.cloud.dataflow.sdk.options.Validation;
+import com.google.cloud.dataflow.sdk.transforms.DoFn;
+import com.google.cloud.dataflow.sdk.transforms.IntraBundleParallelization;
+import com.google.cloud.dataflow.sdk.util.Transport;
+import com.google.common.collect.ImmutableMap;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/**
+ * A batch Dataflow pipeline for injecting a set of GCS files into
+ * a PubSub topic line by line. Empty lines are skipped.
+ *
+ * This is useful for testing streaming
+ * pipelines. Note that since batch pipelines might retry chunks, this
+ * does _not_ guarantee exactly-once injection of file data. Some lines may
+ * be published multiple times.
+ *
+ */
+public class PubsubFileInjector {
+
+ /**
+ * An incomplete {@code PubsubFileInjector} transform with unbound output topic.
+ */
+ public static class Unbound {
+ private final String timestampLabelKey;
+
+ Unbound() {
+ this.timestampLabelKey = null;
+ }
+
+ Unbound(String timestampLabelKey) {
+ this.timestampLabelKey = timestampLabelKey;
+ }
+
+ Unbound withTimestampLabelKey(String timestampLabelKey) {
+ return new Unbound(timestampLabelKey);
+ }
+
+ public Bound publish(String outputTopic) {
+ return new Bound(outputTopic, timestampLabelKey);
+ }
+ }
+
+ /** A DoFn that publishes non-empty lines to Google Cloud PubSub. */
+ public static class Bound extends DoFn {
+ private final String outputTopic;
+ private final String timestampLabelKey;
+ public transient Pubsub pubsub;
+
+ public Bound(String outputTopic, String timestampLabelKey) {
+ this.outputTopic = outputTopic;
+ this.timestampLabelKey = timestampLabelKey;
+ }
+
+ @Override
+ public void startBundle(Context context) {
+ this.pubsub =
+ Transport.newPubsubClient(context.getPipelineOptions().as(DataflowPipelineOptions.class))
+ .build();
+ }
+
+ @Override
+ public void processElement(ProcessContext c) throws IOException {
+ if (c.element().isEmpty()) {
+ return;
+ }
+ PubsubMessage pubsubMessage = new PubsubMessage();
+ pubsubMessage.encodeData(c.element().getBytes());
+ if (timestampLabelKey != null) {
+ pubsubMessage.setAttributes(
+ ImmutableMap.of(timestampLabelKey, Long.toString(c.timestamp().getMillis())));
+ }
+ PublishRequest publishRequest = new PublishRequest();
+ publishRequest.setMessages(Arrays.asList(pubsubMessage));
+ this.pubsub.projects().topics().publish(outputTopic, publishRequest).execute();
+ }
+ }
+
+ /**
+ * Creates a {@code PubsubFileInjector} transform with the given timestamp label key.
+ */
+ public static Unbound withTimestampLabelKey(String timestampLabelKey) {
+ return new Unbound(timestampLabelKey);
+ }
+
+ /**
+ * Creates a {@code PubsubFileInjector} transform that publishes to the given output topic.
+ */
+ public static Bound publish(String outputTopic) {
+ return new Unbound().publish(outputTopic);
+ }
+
+ /**
+ * Command line parameter options.
+ */
+ private interface PubsubFileInjectorOptions extends PipelineOptions {
+ @Description("GCS location of files.")
+ @Validation.Required
+ String getInput();
+ void setInput(String value);
+
+ @Description("Topic to publish on.")
+ @Validation.Required
+ String getOutputTopic();
+ void setOutputTopic(String value);
+ }
+
+ /**
+ * Sets up and starts streaming pipeline.
+ */
+ public static void main(String[] args) {
+ PubsubFileInjectorOptions options = PipelineOptionsFactory.fromArgs(args)
+ .withValidation()
+ .as(PubsubFileInjectorOptions.class);
+
+ Pipeline pipeline = Pipeline.create(options);
+
+ pipeline
+ .apply(TextIO.Read.from(options.getInput()))
+ .apply(IntraBundleParallelization.of(PubsubFileInjector.publish(options.getOutputTopic()))
+ .withMaxParallelism(20));
+
+ pipeline.run();
+ }
+}
diff --git a/examples/src/main/java/com/google/cloud/dataflow/examples/complete/AutoComplete.java b/examples/src/main/java/com/google/cloud/dataflow/examples/complete/AutoComplete.java
new file mode 100644
index 000000000000..1bccc4ace278
--- /dev/null
+++ b/examples/src/main/java/com/google/cloud/dataflow/examples/complete/AutoComplete.java
@@ -0,0 +1,510 @@
+/*
+ * Copyright (C) 2015 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.dataflow.examples.complete;
+
+import com.google.api.services.bigquery.model.TableFieldSchema;
+import com.google.api.services.bigquery.model.TableReference;
+import com.google.api.services.bigquery.model.TableRow;
+import com.google.api.services.bigquery.model.TableSchema;
+import com.google.api.services.datastore.DatastoreV1.Entity;
+import com.google.api.services.datastore.DatastoreV1.Key;
+import com.google.api.services.datastore.DatastoreV1.Value;
+import com.google.api.services.datastore.client.DatastoreHelper;
+import com.google.cloud.dataflow.examples.common.DataflowExampleUtils;
+import com.google.cloud.dataflow.examples.common.ExampleBigQueryTableOptions;
+import com.google.cloud.dataflow.examples.common.ExamplePubsubTopicOptions;
+import com.google.cloud.dataflow.sdk.Pipeline;
+import com.google.cloud.dataflow.sdk.PipelineResult;
+import com.google.cloud.dataflow.sdk.coders.AvroCoder;
+import com.google.cloud.dataflow.sdk.coders.DefaultCoder;
+import com.google.cloud.dataflow.sdk.io.BigQueryIO;
+import com.google.cloud.dataflow.sdk.io.DatastoreIO;
+import com.google.cloud.dataflow.sdk.io.PubsubIO;
+import com.google.cloud.dataflow.sdk.io.TextIO;
+import com.google.cloud.dataflow.sdk.options.Default;
+import com.google.cloud.dataflow.sdk.options.Description;
+import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
+import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
+import com.google.cloud.dataflow.sdk.transforms.Count;
+import com.google.cloud.dataflow.sdk.transforms.DoFn;
+import com.google.cloud.dataflow.sdk.transforms.Filter;
+import com.google.cloud.dataflow.sdk.transforms.Flatten;
+import com.google.cloud.dataflow.sdk.transforms.PTransform;
+import com.google.cloud.dataflow.sdk.transforms.ParDo;
+import com.google.cloud.dataflow.sdk.transforms.Partition;
+import com.google.cloud.dataflow.sdk.transforms.Partition.PartitionFn;
+import com.google.cloud.dataflow.sdk.transforms.SerializableFunction;
+import com.google.cloud.dataflow.sdk.transforms.Top;
+import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindows;
+import com.google.cloud.dataflow.sdk.transforms.windowing.SlidingWindows;
+import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
+import com.google.cloud.dataflow.sdk.transforms.windowing.WindowFn;
+import com.google.cloud.dataflow.sdk.values.KV;
+import com.google.cloud.dataflow.sdk.values.PBegin;
+import com.google.cloud.dataflow.sdk.values.PCollection;
+import com.google.cloud.dataflow.sdk.values.PCollectionList;
+import com.google.common.base.Preconditions;
+
+import org.joda.time.Duration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * An example that computes the most popular hash tags
+ * for every prefix, which can be used for auto-completion.
+ *
+ * Concepts: Using the same pipeline in both streaming and batch, combiners,
+ * composite transforms.
+ *
+ *
To execute this pipeline using the Dataflow service in batch mode,
+ * specify pipeline configuration:
+ *
{@code
+ * --project=YOUR_PROJECT_ID
+ * --stagingLocation=gs://YOUR_STAGING_DIRECTORY
+ * --runner=DataflowPipelineRunner
+ * --inputFile=gs://path/to/input*.txt
+ * }
+ *
+ * To execute this pipeline using the Dataflow service in streaming mode,
+ * specify pipeline configuration:
+ *
{@code
+ * --project=YOUR_PROJECT_ID
+ * --stagingLocation=gs://YOUR_STAGING_DIRECTORY
+ * --runner=DataflowPipelineRunner
+ * --inputFile=gs://YOUR_INPUT_DIRECTORY/*.txt
+ * --streaming
+ * }
+ *
+ * This will update the datastore every 10 seconds based on the last
+ * 30 minutes of data received.
+ */
+public class AutoComplete {
+
+ /**
+ * A PTransform that takes as input a list of tokens and returns
+ * the most common tokens per prefix.
+ */
+ public static class ComputeTopCompletions
+ extends PTransform, PCollection>>> {
+ private final int candidatesPerPrefix;
+ private final boolean recursive;
+
+ protected ComputeTopCompletions(int candidatesPerPrefix, boolean recursive) {
+ this.candidatesPerPrefix = candidatesPerPrefix;
+ this.recursive = recursive;
+ }
+
+ public static ComputeTopCompletions top(int candidatesPerPrefix, boolean recursive) {
+ return new ComputeTopCompletions(candidatesPerPrefix, recursive);
+ }
+
+ @Override
+ public PCollection>> apply(PCollection input) {
+ PCollection