Workflow using Cloud Composer

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.

    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.

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