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

1. Install the Cloud SDK.

   Note: This is not required in Cloud Shell since it already has the Cloud SDK pre-installed.

2. Create a new Google Cloud project via the New Project page, or via the gcloud command line tool.

   export PROJECT_ID=your-google-cloud-project-id
   gcloud projects create $PROJECT_ID

3. Enable billing.

4. Setup the Cloud SDK to your GCP project.

   gcloud init

5. Enable the APIs: Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Cloud Storage JSON, Pub/Sub, Cloud Scheduler, Cloud Resource Manager, and App Engine.

6. Create a service account JSON key via the Create service account key page, or via the gcloud command line tool. Here is how to do it through the Create service account key page.

   • 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 (*).
   • Click Create. A JSON file that contains your key downloads to your computer.

   Alternatively, you can use gcloud through the command line.

   export PROJECT_ID=$(gcloud config get-value project)
   export SERVICE_ACCOUNT_NAME=samples
   export IAM_ACCOUNT=$SERVICE_ACCOUNT_NAME@$PROJECT_ID.iam.gserviceaccount.com
   
   # Create the service account.
   gcloud iam service-accounts create $SERVICE_ACCOUNT_NAME \
     --display-name $SERVICE_ACCOUNT_NAME
   
   # Set the role to Project Owner (*).
   gcloud projects add-iam-policy-binding $PROJECT_ID \
     --member serviceAccount:$IAM_ACCOUNT \
     --role roles/owner
   
   # Create a JSON file with the service account credentials.
   gcloud iam service-accounts keys create path/to/your/credentials.json \
     --iam-account=$IAM_ACCOUNT

   (*) 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.

   For more information, see Creating and managing service accounts.

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

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

8. Create a Cloud Storage bucket.

   export BUCKET_ID=your-gcs-bucket-id
   
   gcloud storage buckets create gs://$BUCKET_ID

9. 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.

   export TOPIC_ID=your-topic-id
   
   # Create a Pub/Sub topic.
   gcloud pubsub topics create $TOPIC_ID
   
   # 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. Install Python and virtualenv.

2. Clone the python-docs-samples repository.

   git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git

3. Navigate to the sample code directory.

   cd python-docs-samples/pubsub/streaming-analytics

4. Create a virtual environment and activate it.

virtualenv env
source env/bin/activate

Once you are finished with the tutorial, you can deactivate the virtualenv and go back to your global Python environment by running deactivate.

1. Install the sample requirements.

pip install -U -r requirements.txt

• PubSubToGCS.py

The following example will run a streaming pipeline. The pipeline does the following:

1. Reads messages from a Pub/Sub topic.
2. Group messages into batches for every windows.
3. Adds window start and end time to each element/message.
4. Adds publish timestamp to each element/message.
5. Adds a random shard ID as key to each windowed element. Sharding lets you split the elements in the same window into multiple small batches. This way, multiple workers can each write a batch of elements into Cloud Storage. This results in one file per shard.
6. Groups the elements by their shard ID for every window.
7. Writes the grouped elements to a file on Cloud Storage.

• --project: sets the Google Cloud project ID to run the pipeline on
• --region: sets the Dataflow regional endpoint
• --input_topic: sets the input Pub/Sub topic to read messages from
• --output_path: sets the output GCS path prefix to write files to
• --runner: specifies the runner to run the pipeline, if not set to DataflowRunner, DirectRunner is used
• --window_size [optional]: specifies the window size in minutes, defaults to 1.0
• --num_shards [optional]: sets the number of shards when writing windowed elements to GCS, defaults to 5.
• --temp_location: needed for executing the pipeline

python PubSubToGCS.py \
  --project=$PROJECT_ID \
  --region=us-central1 \
  --input_topic=projects/$PROJECT_ID/topics/$TOPIC_ID \
  --output_path=gs://$BUCKET_ID/samples/output \
  --runner=DataflowRunner \
  --window_size=1 \
  # If set, you will write up to `num_shards` files per window to GCS.
  # --num_shards=2 \
  --temp_location=gs://$BUCKET_ID/temp

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.

gcloud storage ls gs://$BUCKET_ID/samples/

1. Delete the Google Cloud Scheduler job.

   gcloud scheduler jobs delete publisher-job

2. Ctrl+C to stop the program in your terminal. Note that this does not actually stop the job if you use DataflowRunner.

3. If you use DirectRunner, you can skip this step. Stop the Dataflow job in GCP Console Dataflow page. Cancel the job instead of draining it. This may take some minutes.

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

   gcloud pubsub topics delete $TOPIC_ID

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

   # Delete only the files created by this sample.
   gcloud storage rm --recursive --continue-on-error "gs://$BUCKET_ID/samples/output*"
   
   # [optional] Remove the Cloud Storage bucket.
   gcloud storage buckets delete gs://$BUCKET_ID