tutorial to trigger dataflow jobs using cloud scheduler

tutorials/schedule-dataflow-jobs-with-cloud-scheduler/index.md

@@ -0,0 +1,153 @@
+---
+title: Schedule Your Dataflow Batch Jobs With Cloud Scheduler
+description: Learn how to set up Cloud Scheduler to trigger your Dataflow batch jobs.
+author: zhongchen
+tags: Cloud Dataflow, Cloud Scheduler 
+date_published: 2020-08-11
+---
+
+## Notes
+- It is feasible to trigger a Dataflow batch job directly from the cloud scheduler directly. It is easy and fast. There is no need to use cloud function for that.
+- Cloud schedulers need to be created in the same region of App engine. In your [Terraform script](https://www.terraform.io/docs/providers/google/r/cloud_scheduler_job.html#region), 
+make sure assigning the right value for the region field. You need to use **us-central1** if your app engine lives in **us-central**.
+- Use the [regional endpoint](https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.locations.jobs/create) to specify the region of Dataflow job. 
+If you don't explicitly set the location in the request, the jobs will be created in the default region (US-central).
+
+
+In this tutorial, you will learn how to set up a [Cloud Scheduler](https://cloud.google.com/scheduler/) job to trigger to your 
+Dataflow batch jobs.
+
+
+Here is a high level architecture diagram, and you can find all the code [here](./scheduler-dataflow-demo)..
+![diagram](scheduler-dataflow-diagram.png) 
+
+[Cloud Dataflow](https://cloud.google.com/dataflow) is a managed service for handling 
+both streaming and batch jobs. For your streaming jobs, you just need to launch them once without worrying about operating them afterwards. 
+However, for your batch jobs, you probably need to trigger them based on certain conditions.
+
+First things first, to be able to run your Dataflow jobs on a regular basis, you need to build your Dataflow templates. 
+Follow the [instructions](https://cloud.google.com/dataflow/docs/guides/templates/creating-templates) to create your templates and save them in a GCS bucket.
+
+![Upload Dataflow templates in a GCS bucket](store_a_template_in_gcs.png)
+
+
+Once you have your templates ready, you can set up cloud schedulers to trigger Dataflow templates. 
+Here is one example to define a scheduler using Terraform.
+
+
+```hcl-terraform
+resource "google_cloud_scheduler_job" "scheduler" {
+  name = "scheduler-demo"
+  schedule = "0 0 * * *"
+  # This needs to be us-central1 even if the app engine is in us-central.
+  # You will get a resource not found error if just using us-central.
+  region = "us-central1"
+
+  http_target {
+    http_method = "POST"
+    uri = "https://dataflow.googleapis.com/v1b3/projects/${var.project_id}/locations/${var.region}/templates:launch?gcsPath=gs://${var.bucket}/templates/dataflow-demo-template"
+    oauth_token {
+      service_account_email = google_service_account.cloud-scheduler-demo.email
+    }
+
+    # need to encode the string
+    body = base64encode(<<-EOT
+    {
+      "jobName": "test-cloud-scheduler",
+      "parameters": {
+        "region": "${var.region}",
+        "autoscalingAlgorithm": "THROUGHPUT_BASED",
+      },
+      "environment": {
+        "maxWorkers": "10",
+        "tempLocation": "gs://${var.bucket}/temp",
+        "zone": "${var.region}-a"
+      }
+    }
+EOT
+    )
+  }
+}
+```
+
+## Instructions
+
+The following step-by-step instructions can be used to create a sample Dataflow pipeline with Cloudbuild.
+
+First, open Cloud Shell and clone the repository.
+
+```
+git clone https://github.com/GoogleCloudPlatform/community
+cd community/tutorials/schedule-dataflow-jobs-with-cloud-scheduler/scheduler-dataflow-demo/
+```
+
+Create a bucket on Google Cloud Storage, which will be used to store terraform states and Dataflow templates.
+Replace BUCKET with your own choice. ${GOOGLE_CLOUD_PROJECT} is predefined in Cloud Shell for the project ID.
+You can skip this step if you already have one GCS bucket created.
+
+```
+export BUCKET=[BUCKET]
+gsutil mb -p ${GOOGLE_CLOUD_PROJECT} gs://${BUCKET}
+```
+
+The next step is to create a backend for Terraform to store the states of 
+GCP resources. Run the below command to create a remote backend service using GCS as the storage.
+```
+cd terraform
+cat > backend.tf << EOF
+terraform {
+ backend "gcs" {
+   bucket  = "${BUCKET}"
+   prefix  = "terraform/state"
+ }
+}
+EOF
+```
+
+Follow the [instruction](https://cloud.google.com/scheduler/docs/quickstart) to create a App Engine, which is needed to
+set up Cloud Scheduler jobs.
+
+Note: Cloud Scheduler jobs need to be created in the same region as the App engine. 
+In addition, you need to set the region to be *us-central1* even when the region shows as
+*us-central* on the UI.
+
+![app engine location](app_engine_location.png)
+
+```
+export REGION=us-central1
+```
+
+Since we will be using Cloudbuild to execute GCP resource creation steps, it is
+important to make sure the Cloudbuild service account has all the proper roles.
+Follow the [link](https://cloud.google.com/cloud-build/docs/securing-builds/configure-access-for-cloud-build-service-account#granting_a_role_using_the_iam_page) 
+to enable all the roles needed.
+
+- Cloud Scheduler Admin
+- Dataflow Admin
+- Service Account User
+- Project IAM Admin
+
+Verify that all the roles are enabled in the UI.
+
+![cloudbuild_sa_status](cloudbuild_sa_setup.png)
+
+Now you can submit a Cloudbuild job to create all the resources.
+
+
+```
+cd ..
+gcloud builds submit --config=cloudbuild.yaml \
+  --substitutions=_BUCKET=${BUCKET},_REGION=${REGION},_PROJECT_ID=${GOOGLE_CLOUD_PROJECT} .
+```
+
+The job will run based on the schedule you defined in the terraform script. 
+In addition, you can manually run the scheduler through UI and watch it trigger your Dataflow batch job. 
+You can check the status of jobs through the UI.
+
+![See the status of your jobs](check_scheduler_status.png)
+
+
+
+## Cleaning up
+
+Since this tutorial uses multiple GCP components, please be sure to delete the associated resources once you are done.

tutorials/schedule-dataflow-jobs-with-cloud-scheduler/scheduler-dataflow-demo/README.md

@@ -0,0 +1,3 @@
+# Cloud Scheduler & Dataflow Demo
+
+The folder contains an example how to set up a cloud scheduler to trigger a Dataflow batch job.

tutorials/schedule-dataflow-jobs-with-cloud-scheduler/scheduler-dataflow-demo/cloudbuild.yaml

@@ -0,0 +1,29 @@
+# The PROJECT_ID is a cloud build env variable
+steps:
+- id: 'Terraform init'
+  name: 'hashicorp/terraform:0.12.20'
+  dir: 'terraform'
+  entrypoint: 'sh'
+  args:
+    - '-c'
+    - |
+      terraform init -input=false
+      terraform workspace select scheduler-dataflow-demo || terraform workspace new scheduler-dataflow-demo
+      terraform apply -input=false -var=project_id=${_PROJECT_ID} -var=region=${_REGION}\
+        -var=bucket=${_BUCKET} -auto-approve
+  waitFor: ['-']
+
+- id: "Build dataflow template"
+  name: maven:3.6.0-jdk-11-slim
+  dir: 'dataflow'
+  env:
+    - "PROJECT=${_PROJECT_ID}"
+    - "BUCKET=${_BUCKET}"
+    - "REGION=${_REGION}"
+  entrypoint: 'bash'
+  args:
+    - '-c'
+    - |
+      ./build.sh
+  waitFor: ['Terraform init']
+

tutorials/schedule-dataflow-jobs-with-cloud-scheduler/scheduler-dataflow-demo/dataflow/build.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+mvn compile exec:java \
+           -Dexec.mainClass=DataflowDemoPipeline \
+           -Dexec.args="--runner=DataflowRunner \
+                        --project=${PROJECT} \
+                        --stagingLocation=gs://${BUCKET}/staging \
+                        --gcpTempLocation=gs://${BUCKET}/temp \
+                        --region=${REGION} \
+                        --templateLocation=gs://${BUCKET}/templates/dataflow-demo-template"

tutorials/schedule-dataflow-jobs-with-cloud-scheduler/scheduler-dataflow-demo/dataflow/pom.xml

@@ -0,0 +1,127 @@
+<?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>dataflow</groupId>
+    <artifactId>demo</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+
+    <properties>
+        <beam.version>2.15.0</beam.version>
+        <maven-compiler-plugin.version>3.7.0</maven-compiler-plugin.version>
+        <maven-jar-plugin.version>3.0.2</maven-jar-plugin.version>
+        <slf4j.version>1.7.25</slf4j.version>
+        <jackson.version>2.8.8</jackson.version>
+    </properties>
+    <build>
+    <plugins>
+    <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <version>${maven-compiler-plugin.version}</version>
+        <configuration>
+            <source>1.8</source>
+            <target>1.8</target>
+        </configuration>
+    </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>org.apache.streaming.WordCount</mainClass>
+                    </manifest>
+                </archive>
+            </configuration>
+        </plugin>
+    </plugins>
+    </build>
+
+    <profiles>
+
+        <profile>
+            <id>direct-runner</id>
+            <activation>
+                <activeByDefault>true</activeByDefault>
+            </activation>
+            <!-- Makes the DirectRunner available when running a pipeline. -->
+            <dependencies>
+                <dependency>
+                    <groupId>org.apache.beam</groupId>
+                    <artifactId>beam-runners-direct-java</artifactId>
+                    <version>${beam.version}</version>
+                    <scope>runtime</scope>
+                </dependency>
+            </dependencies>
+    </profile>
+        <profile>
+            <id>dataflow-runner</id>
+            <!-- Makes the DataflowRunner available when running a pipeline. -->
+            <dependencies>
+                <dependency>
+                    <groupId>org.apache.beam</groupId>
+                    <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
+                    <version>${beam.version}</version>
+                    <scope>runtime</scope>
+                </dependency>
+            </dependencies>
+        </profile>
+
+    </profiles>
+
+    <dependencies>
+        <!-- Adds a dependency on the Beam SDK. -->
+        <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-google-cloud-dataflow-java</artifactId>
+            <version>${beam.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>28.1-jre</version>
+        </dependency>
+
+        <!-- The DirectRunner is needed for unit tests. -->
+        <dependency>
+            <groupId>org.apache.beam</groupId>
+            <artifactId>beam-runners-direct-java</artifactId>
+            <version>${beam.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <!-- Adds a dependency on the Beam Google Cloud Platform IO module. -->
+        <dependency>
+            <groupId>org.apache.beam</groupId>
+            <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
+            <version>${beam.version}</version>
+        </dependency>
+
+        <!-- Add 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>
+            <!-- When loaded at runtime this will wire up slf4j to the JUL backend -->
+            <scope>runtime</scope>
+        </dependency>
+    </dependencies>
+</project>

tutorials/schedule-dataflow-jobs-with-cloud-scheduler/scheduler-dataflow-demo/dataflow/src/main/java/DataflowDemoPipeline.java

@@ -0,0 +1,43 @@
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PDone;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class DataflowDemoPipeline {
+    private static final Logger Log = LoggerFactory.getLogger(DataflowDemoPipeline.class);
+
+    public static void main(String[] args) {
+        // Register Options class for our pipeline with the factory
+        PipelineOptionsFactory.register(DemoPipelineOptions.class);
+
+        DemoPipelineOptions options = PipelineOptionsFactory.fromArgs(args)
+                .withValidation()
+                .as(DemoPipelineOptions.class);
+
+        Pipeline p = Pipeline.create(options);
+
+        String[] numbers = new String[]{"one", "two", "three"};
+        List<String> input = Arrays.asList(numbers);
+        p.apply("dummy input", Create.of(input)).
+                apply("dummy transformation", ParDo.of(
+                        new DoFn<String, String>() {
+                            @ProcessElement
+                            public void processElement(ProcessContext context) {
+                                String elem = context.element();
+                                Log.info("Process element: " + elem);
+                                context.output(elem);
+                            }
+                        }
+                ));
+        PDone.in(p);
+        p.run();
+    }
+}
+

tutorials/schedule-dataflow-jobs-with-cloud-scheduler/scheduler-dataflow-demo/dataflow/src/main/java/DemoPipelineOptions.java

@@ -0,0 +1,11 @@
+import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
+import org.apache.beam.sdk.options.Default;
+import org.apache.beam.sdk.options.Description;
+import org.apache.beam.sdk.options.ValueProvider;
+
+public interface DemoPipelineOptions extends DataflowPipelineOptions {
+    @Description("Subscription name")
+    @Default.String("demo_subscription")
+    ValueProvider<String> getSubscription();
+    void setSubscription(ValueProvider<String> subscription);
+}