Sample(s) showing how to use [Google Cloud Pub/Sub] with [Google Cloud Dataflow].

1. Install the [Cloud SDK].

1. [Create a new project].

1. [Enable billing].

1. [Enable the APIs](https://console.cloud.google.com/flows/enableapi?apiid=dataflow,compute_component,logging,storage_component,storage_api,pubsub,cloudresourcemanager.googleapis.com,cloudscheduler.googleapis.com,appengine.googleapis.com): Dataflow, Compute Engine, Stackdriver Logging, Cloud Storage, Cloud Storage JSON, Pub/Sub, Cloud Scheduler, Cloud Resource Manager, and App Engine.

1. Setup the Cloud SDK to your GCP project.

```bash

gcloud init

```

1. [Create a service account key] as a JSON file.

For more information, see [Creating and managing service accounts].

* From the **Service account** list, select **New service account**.

* In the **Service account name** field, enter a name.

* From the **Role** list, select **Project > Owner**.

> **Note**: The **Role** field authorizes your service account to access resources.

> You can view and change this field later by using the [GCP Console IAM page].

> If you are developing a production app, specify more granular permissions than **Project > Owner**.

> For more information, see [Granting roles to service accounts].

* Click **Create**. A JSON file that contains your key downloads to your computer.

1. Set your `GOOGLE_APPLICATION_CREDENTIALS` environment variable to point to your service account key file.

```bash

export GOOGLE_APPLICATION_CREDENTIALS=path/to/your/credentials.json

```

1. Create a Cloud Storage bucket.

```bash

BUCKET_NAME=your-gcs-bucket

PROJECT_NAME=$(gcloud config get-value project)

gsutil mb gs://$BUCKET_NAME

```

1. Start a [Google Cloud Scheduler] job that publishes one message to a [Google Cloud Pub/Sub] topic every minute. This will create an [App Engine] app if one has never been created on the project.

```bash

# Create a Pub/Sub topic.

gcloud pubsub topics create cron-topic

# Create a Cloud Scheduler job

gcloud scheduler jobs create pubsub publisher-job --schedule="* * * * *" \

--topic=cron-topic --message-body="Hello!"

# Run the job.

gcloud scheduler jobs run publisher-job

```

The following instructions will help you prepare your development environment.

1. Download and install the [Java Development Kit (JDK)].

Verify that the [JAVA_HOME] environment variable is set and points to your JDK installation.

1. Download and install [Apache Maven] by following the [Maven installation guide] for your specific operating system.

1. Clone the `java-docs-samples` repository.

```bash

1. Navigate to the sample code directory.

```bash

* [PubSubToGCS.java](src/main/java/com/examples/pubsub/streaming/PubSubToGCS.java)

The following example will run a streaming pipeline. It will read messages from a Pub/Sub topic, then window them into fixed-sized intervals, and write one file per window into a GCS location.

+ `--project`: sets the Google Cloud project ID to run the pipeline on

+ `--inputTopic`: sets the input Pub/Sub topic to read messages from

+ `--output`: sets the output GCS path prefix to write files to

+ `--runner [optional]`: specifies the runner to run the pipeline, defaults to `DirectRunner`

+ `--windowSize [optional]`: specifies the window size in minutes, defaults to 1

After the job has been submitted, you can check its status in the [GCP Console Dataflow page].

You can also check the output to your GCS bucket using the command line below or in the [GCP Console Storage page]. You may need to wait a few minutes for the files to appear.

1. Delete the [Google Cloud Scheduler] job.

```bash

1. `Ctrl+C` to stop the program in your terminal. Note that this does not actually stop the job if you use `DataflowRunner`. Skip 3 if you use the `DirectRunner`.

1. Stop the Dataflow job in [GCP Console Dataflow page]. Cancel the job instead of draining it. This may take some minutes.

1. Delete the topic. [Google Cloud Dataflow] will automatically delete the subscription associated with the streaming pipeline when the job is canceled.

```bash

1. Lastly, to avoid incurring charges to your GCP account for the resources created in this tutorial:

```bash

[Apache Beam]: https://beam.apache.org/

[Google Cloud Pub/Sub]: https://cloud.google.com/pubsub/docs/

[Google Cloud Dataflow]: https://cloud.google.com/dataflow/docs/

[Google Cloud Scheduler]: https://cloud.google.com/scheduler/docs/

[App Engine]: https://cloud.google.com/appengine/docs/

[Cloud SDK]: https://cloud.google.com/sdk/docs/

[Create a new project]: https://console.cloud.google.com/projectcreate

[Enable billing]: https://cloud.google.com/billing/docs/how-to/modify-project

[Create a service account key]: https://console.cloud.google.com/apis/credentials/serviceaccountkey

[Creating and managing service accounts]: https://cloud.google.com/iam/docs/creating-managing-service-accounts

[GCP Console IAM page]: https://console.cloud.google.com/iam-admin/iam

[Granting roles to service accounts]: https://cloud.google.com/iam/docs/granting-roles-to-service-accounts

[Java Development Kit (JDK)]: https://www.oracle.com/technetwork/java/javase/downloads/index.html

[JAVA_HOME]: https://docs.oracle.com/javase/8/docs/technotes/guides/troubleshoot/envvars001.html

[Apache Maven]: http://maven.apache.org/download.cgi

[Maven installation guide]: http://maven.apache.org/install.html

[GCP Console create Dataflow job page]: https://console.cloud.google.com/dataflow/createjob

[GCP Console Dataflow page]: https://console.cloud.google.com/dataflow

[GCP Console Storage page]: https://console.cloud.google.com/storage

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

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

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

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

<modelVersion>4.0.0</modelVersion>

<groupId>com.example</groupId>

<artifactId>pubsub-streaming</artifactId>

<version>1.0</version>

<properties>

<maven.compiler.source>1.8</maven.compiler.source>

<maven.compiler.target>1.8</maven.compiler.target>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

<beam.version>2.14.0</beam.version>

<maven-compiler-plugin.version>3.8.1</maven-compiler-plugin.version>

<maven-exec-plugin.version>1.6.0</maven-exec-plugin.version>

<maven-jar-plugin.version>3.1.2</maven-jar-plugin.version>

<maven-shade-plugin.version>3.2.1</maven-shade-plugin.version>

<slf4j.version>1.7.28</slf4j.version>

</properties>

<repositories>

<repository>

<id>apache.snapshots</id>

<name>Apache Development Snapshot Repository</name>

<url>https://repository.apache.org/content/repositories/snapshots/</url>

<releases>

<enabled>false</enabled>

</releases>

<snapshots>

<enabled>true</enabled>

</snapshots>

</repository>

</repositories>

<build>

<plugins>

<plugin>

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

<artifactId>maven-compiler-plugin</artifactId>

<version>${maven-compiler-plugin.version}</version>

</plugin>

<plugin>

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

<artifactId>maven-jar-plugin</artifactId>

<version>${maven-jar-plugin.version}</version>

<configuration>

<archive>

<manifest>

<addClasspath>true</addClasspath>

<classpathPrefix>lib/</classpathPrefix>

<mainClass>com.examples.pubsub.streaming.PubSubToGCS</mainClass>

</manifest>

</archive>

</configuration>

</plugin>

<!--

<plugin>

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

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

<version>${maven-shade-plugin.version}</version>

<executions>

<execution>

<phase>package</phase>

<goals>

<goal>shade</goal>

</goals>

<configuration>

<finalName>${project.artifactId}-bundled-${project.version}</finalName>

<filters>

<filter>

<artifact>*:*</artifact>

<excludes>

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

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

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

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

</excludes>

</filter>

</filters>

<transformers>

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

</transformers>

</configuration>

</execution>

</executions>

</plugin>

</plugins>

<pluginManagement>

<plugins>

<plugin>

<groupId>org.codehaus.mojo</groupId>

<artifactId>exec-maven-plugin</artifactId>

<version>${maven-exec-plugin.version}</version>

<configuration>

<cleanupDaemonThreads>false</cleanupDaemonThreads>

</configuration>

</plugin>

</plugins>

</pluginManagement>

</build>

<dependencies>

<dependency>

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

<artifactId>beam-sdks-java-core</artifactId>

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

</dependency>

<!--

<dependency>

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

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

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

<scope>runtime</scope>

</dependency>

<dependency>

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

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

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

<scope>runtime</scope>

</dependency>

<dependency>

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

<artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>

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

</dependency>

<dependency>

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

<artifactId>beam-sdks-java-maven-archetypes-examples</artifactId>

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

</dependency>

<!-- slf4j API frontend binding with JUL backend -->

<dependency>

<groupId>org.slf4j</groupId>

<artifactId>slf4j-api</artifactId>

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

</dependency>

<dependency>

<groupId>org.slf4j</groupId>

<artifactId>slf4j-jdk14</artifactId>

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

</dependency>

</dependencies>

</project>