Workflow using Cloud Composer

Objective:

1. Create a Dataproc workflow template that runs a Spark PI job
2. Create an Apache Airflow DAG that Cloud Composer will use to start the workflow at a specific time.

In this document, you use the following billable components of Google Cloud:

• Dataproc
• Compute Engine
• Cloud Composer

To generate a cost estimate based on your projected usage, use the pricing calculator. New Google Cloud users might be eligible for a free trial.

Before you begin

Set up your project

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

3. Make sure that billing is enabled for your Google Cloud project.

4. Enable the Dataproc, Compute Engine, and Cloud Composer APIs.

   Enable the APIs

5. Install the Google Cloud CLI.

6. To initialize the gcloud CLI, run the following command:

   gcloud init

7. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

8. Make sure that billing is enabled for your Google Cloud project.

9. Enable the Dataproc, Compute Engine, and Cloud Composer APIs.

   Enable the APIs

10. Install the Google Cloud CLI.

11. To initialize the gcloud CLI, run the following command:

    gcloud init

Create a Dataproc workflow template

Copy and run the commands listed below in a local terminal window or in Cloud Shell to create and define a workflow template.

1. Create the sparkpi workflow template.

   gcloud dataproc workflow-templates create sparkpi \
       --region=us-central1
         

2. Add the spark job to the sparkpi workflow template. The "compute" step-id flag identifies the SparkPi job.

   gcloud dataproc workflow-templates add-job spark \
       --workflow-template=sparkpi \
       --step-id=compute \
       --class=org.apache.spark.examples.SparkPi \
       --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
       --region=us-central1 \
       -- 1000
         

3. Use a managed, single-node cluster to run the workflow. Dataproc will create the cluster, run the workflow on it, then delete the cluster when the workflow completes.

   gcloud dataproc workflow-templates set-managed-cluster sparkpi \
       --cluster-name=sparkpi \
       --single-node \
       --region=us-central1
         

4. Confirm workflow template creation.

   Console

   Click on the sparkpi name on the Dataproc Workflows page in the Google Cloud console to open the Workflow template details page. Click on the name of your workflow template to confirm the sparkpi template attributes.

   

   gcloud command

   Run the following command:

   gcloud dataproc workflow-templates describe sparkpi --region=us-central1
       

Create and Upload a DAG to Cloud Storage

1. Create or use an existing Cloud Composer environment.
2. Set environment variables.

   Airflow UI

   1. In the toolbar, click Admin > Variables.

      

   2. Click Create.

      

   3. Enter the following information:
      • Key:project_id
      • Val: PROJECT_ID — your Google Cloud Project ID
   4. Click Save.

   gcloud command

   Enter the following commands:

   • ENVIRONMENT is the name of the Cloud Composer environment
   • LOCATION is the region where the Cloud Composer environment is located
   • PROJECT_ID is the project ID for the project that contains the Cloud Composer environment

       gcloud composer environments run ENVIRONMENT --location LOCATION variables set -- project_id PROJECT_ID
       

3. Copy the following DAG code locally into a file titled "composer-dataproc-dag.py", which uses the DataprocInstantiateWorkflowTemplateOperator.

   Airflow 2

   
   """Example Airflow DAG that kicks off a Cloud Dataproc Template that runs a
   Spark Pi Job.
   
   This DAG relies on an Airflow variable
   https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
   * project_id - Google Cloud Project ID to use for the Cloud Dataproc Template.
   """
   
   import datetime
   
   from airflow import models
   from airflow.providers.google.cloud.operators.dataproc import (
       DataprocInstantiateWorkflowTemplateOperator,
   )
   from airflow.utils.dates import days_ago
   
   project_id = "{{var.value.project_id}}"
   
   
   default_args = {
       # Tell airflow to start one day ago, so that it runs as soon as you upload it
       "start_date": days_ago(1),
       "project_id": project_id,
   }
   
   # Define a DAG (directed acyclic graph) of tasks.
   # Any task you create within the context manager is automatically added to the
   # DAG object.
   with models.DAG(
       # The id you will see in the DAG airflow page
       "dataproc_workflow_dag",
       default_args=default_args,
       # The interval with which to schedule the DAG
       schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
   ) as dag:
       start_template_job = DataprocInstantiateWorkflowTemplateOperator(
           # The task id of your job
           task_id="dataproc_workflow_dag",
           # The template id of your workflow
           template_id="sparkpi",
           project_id=project_id,
           # The region for the template
           region="us-central1",
       )
   

   Airflow 1

   
   """Example Airflow DAG that kicks off a Cloud Dataproc Template that runs a
   Spark Pi Job.
   
   This DAG relies on an Airflow variable
   https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
   * project_id - Google Cloud Project ID to use for the Cloud Dataproc Template.
   """
   
   import datetime
   
   from airflow import models
   from airflow.contrib.operators import dataproc_operator
   from airflow.utils.dates import days_ago
   
   project_id = "{{var.value.project_id}}"
   
   
   default_args = {
       # Tell airflow to start one day ago, so that it runs as soon as you upload it
       "start_date": days_ago(1),
       "project_id": project_id,
   }
   
   # Define a DAG (directed acyclic graph) of tasks.
   # Any task you create within the context manager is automatically added to the
   # DAG object.
   with models.DAG(
       # The id you will see in the DAG airflow page
       "dataproc_workflow_dag",
       default_args=default_args,
       # The interval with which to schedule the DAG
       schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
   ) as dag:
       start_template_job = dataproc_operator.DataprocWorkflowTemplateInstantiateOperator(
           # The task id of your job
           task_id="dataproc_workflow_dag",
           # The template id of your workflow
           template_id="sparkpi",
           project_id=project_id,
           # The region for the template
           # For more info on regions where Dataflow is available see:
           # https://cloud.google.com/dataflow/docs/resources/locations
           region="us-central1",
       )
   

4. Upload your DAG to your environment folder in Cloud Storage. After the upload has been completed successfully, click on the DAGs Folder link on the Cloud Composer Environment's page.

   

Viewing a task's status

Airflow UI

1. Open the Airflow web interface.
2. On the DAGs page, click the DAG name (for example, dataproc_workflow_dag).
3. On the DAGs Details page, click Graph View.
4. Check status:
   • Failed: The task has a red box around it. You can also hold the pointer over task and look for State: Failed.
   • Success: The task has a green box around it. You can also hold the pointer over the task and check for State: Success.

Console

Click the Workflows tab to see workflow status.

gcloud command

gcloud dataproc operations list \
    --region=us-central1 \
    --filter="labels.goog-dataproc-workflow-template-id=sparkpi"
    

Cleaning up

To avoid incurring charges to your Google Cloud account, you can delete the resources used in this tutorial:

1. Delete the Cloud Composer environment.

2. Delete the workflow template.

What's next

• See Overview of Dataproc Workflow Templates
• See Workflow scheduling solutions