Workflow using Cloud Composer

This tutorial uses 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 Account.

    If you don't already have one, sign up for a new account.

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

    Go to the project selector page

  3. Make sure that billing is enabled for your Google Cloud project. Learn how to confirm billing is enabled for your project.

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

    Enable the APIs

  5. Install and initialize the Cloud SDK.

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 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
    gcloud composer environments run ENVIRONMENT \
        --location LOCATION
        
  3. Copy the following DAG code locally into a file titled "composer-dataproc-dag.py", which uses the DataprocWorkflowTemplateInstantiateOperator.

    
    """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/concepts.html#variables
    * 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 = models.Variable.get("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
            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. the task has a red box around it, indicating it has failed
    • Success: The task has a green box around it. You can also hold the pointer over the task and check for State: Success. the task has a green box around it, indicating it has succeeded

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