Running a Hadoop wordcount job on a Dataproc cluster

This tutorial shows how to use Cloud Composer to create an Apache Airflow DAG (workflow) that runs an Apache Hadoop wordcount job on a Dataproc cluster using the Google Cloud Console.

Objectives

1. Access your Cloud Composer environment and use the Airflow web interface.
2. Create and view Airflow environment variables.
3. Create and run a DAG that includes the following tasks:
   1. Creates a Dataproc cluster
   2. Runs an Apache Hadoop word-count job on the cluster
   3. Outputs the word-count results to a Cloud Storage bucket
   4. Deletes the cluster

Costs

This tutorial uses the billable components of Google Cloud, including:

• Cloud Composer
• Dataproc
• Cloud Storage

Note: The quotas or limits that apply to the Google Cloud products that Cloud Composer uses for your Airflow deployment, apply to your environment.

It takes up to 25 minutes for the system to create your environment. This tutorial can take approximately 1 hour to complete. Use the pricing calculator to generate a cost estimate based on your projected usage. New Google Cloud users might be eligible for a free trial.

Before you begin

1. Google アカウントにログインします。

   Google アカウントをまだお持ちでない場合は、新しいアカウントを登録します。

2. GCP Console のプロジェクト セレクタのページで、GCP プロジェクトを選択または作成します。

   注: この手順で作成するリソースをそのまま保持する予定でない場合、既存のプロジェクトを選択するのではなく、プロジェクトを作成してください。チュートリアルの完了後にそのプロジェクトを削除すれば、プロジェクトに関連するすべてのリソースを削除できます。

   プロジェクト セレクタのページに移動

3. Google Cloud Platform プロジェクトに対して課金が有効になっていることを確認します。 プロジェクトに対して課金が有効になっていることを確認する方法を学習する。

4. Cloud Composer, Cloud Dataproc, and Cloud Storage必要な API を有効にします。

   API を有効にする

5. In your project, create a Cloud Storage bucket of any storage class and region to store the results of the Hadoop word-count job.
   For data residency requirements or performance benefits, create the storage bucket in the same region you plan to create your environment in.
6. Note the path of the bucket that you created, for example gs://my-bucket. You'll define an Airflow variable for this path and use the variable in the example DAG.

Creating an environment

1. In the Cloud Console, go to the Create environment page.

   Open the Create environment page

2. In the Name field, enter example-environment.

3. In the Location drop-down list, select a region for the Cloud Composer environment. See Available regions for information on selecting a region.

4. For other environment configuration options, use the provided defaults.

5. To create the environment, click Create.

6. Wait until environment creation is completed. When done, the green check mark displays to the left of the environment name.

Viewing environment details

After environment creation is completed, you can view your environment's deployment information, such as the Cloud Composer and Python versions, the URL for the Airflow web interface, and the Google Kubernetes Engine cluster ID.

To view deployment information:

1. In the Cloud Console, go to the Environments page.

   Open the Environments page

2. To view Environment details page, click example-environment.

   

3. Note the zone in which you created your environment, for example us-central-1c. You'll define an Airflow variable for this zone and use it in the example DAG.

Setting Airflow variables

Airflow variables are an Airflow-specific concept that is distinct from environment variables. In this step, you'll use the Airflow web interface to set three Airflow variables to use later in the example DAG.

To set variables:

1. Access the Airflow web interface in Cloud Console:

   1. In the Cloud Console, go to the Environments page.

      Open the Environments page

   2. In the Airflow webserver column for example-environment, click the Airflow link. The Airflow web interface opens in a new window.

2. Set variables in the Airflow web interface:

   1. In the toolbar, click Admin > Variables.
   2. To create a new variable, click Create.
   3. For each of the following variables, enter the Key-Value pair and click Save. All Airflow variables display on the List tab.

      KEY	VALUE
      gcp_project	The Google Cloud Platform project you're using for this tutorial, such as composer-test.
      gcs_bucket	The Cloud Storage bucket you created for this tutorial, such as gs://my-bucket.
      gce_zone	The Compute Engine zone for your environment, such as us-central1-c. This is the zone where your Dataproc cluster will be created. See Available regions and zones.

If you install the Cloud SDK or use Cloud Shell, you can use the gcloud command-line tool to execute Airflow CLI sub-commands. The following example gcloud composer command with the variables sub-command sets the Airflow variable gcp_project to project-id:

gcloud composer environments run environment-name --location us-central1 variables -- --set gcp_project project-id.

For information about using gcloud commands, see Using the Cloud SDK.

Viewing the example workflow

An Airflow DAG is a collection of organized tasks that you want to schedule and run. DAGs are defined in standard Python files. The code shown in hadoop_tutorial.py is the workflow code.

composer/workflows/hadoop_tutorial.py

View on GitHub

"""Example Airflow DAG that creates a Cloud Dataproc cluster, runs the Hadoop
wordcount example, and deletes the cluster.

This DAG relies on three Airflow variables
https://airflow.apache.org/concepts.html#variables
* gcp_project - Google Cloud Project to use for the Cloud Dataproc cluster.
* gce_zone - Google Compute Engine zone where Cloud Dataproc cluster should be
  created.
* gcs_bucket - Google Cloud Storage bucket to use for result of Hadoop job.
  See https://cloud.google.com/storage/docs/creating-buckets for creating a
  bucket.
"""

import datetime
import os

from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils import trigger_rule

# Output file for Cloud Dataproc job.
output_file = os.path.join(
    models.Variable.get('gcs_bucket'), 'wordcount',
    datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep
# Path to Hadoop wordcount example available on every Dataproc cluster.
WORDCOUNT_JAR = (
    'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar'
)
# Arguments to pass to Cloud Dataproc job.
wordcount_args = ['wordcount', 'gs://pub/shakespeare/rose.txt', output_file]

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1),
    datetime.datetime.min.time())

default_dag_args = {
    # Setting start date as yesterday starts the DAG immediately when it is
    # detected in the Cloud Storage bucket.
    'start_date': yesterday,
    # To email on failure or retry set 'email' arg to your email and enable
    # emailing here.
    'email_on_failure': False,
    'email_on_retry': False,
    # If a task fails, retry it once after waiting at least 5 minutes
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': models.Variable.get('gcp_project')
}

with models.DAG(
        'composer_hadoop_tutorial',
        # Continue to run DAG once per day
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:

    # Create a Cloud Dataproc cluster.
    create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
        task_id='create_dataproc_cluster',
        # Give the cluster a unique name by appending the date scheduled.
        # See https://airflow.apache.org/code.html#default-variables
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        num_workers=2,
        zone=models.Variable.get('gce_zone'),
        master_machine_type='n1-standard-1',
        worker_machine_type='n1-standard-1')

    # Run the Hadoop wordcount example installed on the Cloud Dataproc cluster
    # master node.
    run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator(
        task_id='run_dataproc_hadoop',
        main_jar=WORDCOUNT_JAR,
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        arguments=wordcount_args)

    # Delete Cloud Dataproc cluster.
    delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator(
        task_id='delete_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        # Setting trigger_rule to ALL_DONE causes the cluster to be deleted
        # even if the Dataproc job fails.
        trigger_rule=trigger_rule.TriggerRule.ALL_DONE)

    # Define DAG dependencies.
    create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

Operators

An operator is a template for a single task in a workflow. To orchestrate the three tasks in the example workflow, the DAG imports the following three operators:

1. DataprocClusterCreateOperator: Creates a Dataproc cluster.
2. DataProcHadoopOperator: Submits a Hadoop wordcount job and writes results to a Cloud Storage bucket.
3. DataprocClusterDeleteOperator: Deletes the cluster to avoid incurring ongoing Compute Engine charges.

Dependencies

You organize tasks that you want to run in a way that reflects their relationships and dependencies. The tasks in this DAG run sequentially. In this example, the relationship is set in the direction that the Python bitshift operator points (>>).

composer/workflows/hadoop_tutorial.py

View on GitHub

# Define DAG dependencies.
create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

Scheduling

The name of the DAG is composer_hadoop_tutorial, and the DAG runs once each day. Because the start_date that is passed in to default_dag_args is set to yesterday, Cloud Composer schedules the workflow to start immediately after the DAG uploads.

composer/workflows/hadoop_tutorial.py

View on GitHub

with models.DAG(
        'composer_hadoop_tutorial',
        # Continue to run DAG once per day
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:

Uploading the DAG to Cloud Storage

Cloud Composer schedules only the DAGs in the DAGs folder. The DAGs folder is in the Cloud Storage bucket that Cloud Composer creates automatically for your environment.

To upload the DAG:

1. On your local machine, save hadoop_tutorial.py.

2. In the Cloud Console, go to the Environments page.

   Open the Environments page

3. In the DAGs folder column for example-environment, click the DAGs link. The DAGs folder in Cloud Storage opens.

4. Click Upload files.

5. Select hadoop_tutorial.py on your local machine and click Open.

Cloud Composer adds the DAG to Airflow and schedules the DAG automatically. DAG changes occur within 3-5 minutes.

Exploring DAG runs

Viewing task status

When you upload your DAG file to the dags/ folder in Cloud Storage, Cloud Composer parses the file. When completed successfully, the name of the workflow appears in the DAG listing, and the workflow is queued to run immediately.

1. To see task status, go to the Airflow web interface and click DAGs in the toolbar.

   

2. To open the DAG details page, click composer_hadoop_tutorial. This page includes a graphical representation of workflow tasks and dependencies.

   

3. To see each task's status, click Graph View and then mouseover the graphic for each task.

   

Queuing the workflow again

To run the workflow again from the Graph View:

1. In the Airflow UI Graph View, click the create_dataproc_cluster graphic.
2. To reset the three tasks, click Clear and then click OK to confirm.
3. Click create_dataproc_cluster again in Graph View.
4. To queue the workflow again, click Run.

Viewing task results

You can also check the status and results of the composer_hadoop_tutorial workflow by going to the following Cloud Console pages:

• Dataproc Clusters to monitor cluster creation and deletion. Note that the cluster created by the workflow is ephemeral: it only exists for the duration of the workflow and is deleted as part of the last workflow task.

• Dataproc Jobs to view or monitor the Apache Hadoop wordcount job. Click the Job ID to see job log output.

• Cloud Storage Browser to see the results of the wordcount in the wordcount folder in the Cloud Storage bucket you created for this tutorial.

Cleaning up

To avoid incurring charges to your Google Cloud Platform account for the resources used in this tutorial:

Caution: Deleting a project has the following effects:

• Everything in the project is deleted. If you used an existing project for this tutorial, when you delete it, you also delete any other work you've done in the project.
• Custom project IDs are lost. When you created this project, you might have created a custom project ID that you want to use in the future. To preserve the URLs that use the project ID, such as an appspot.com URL, delete selected resources inside the project instead of deleting the whole project.

If you plan to explore multiple tutorials and quickstarts, reusing projects can help you avoid exceeding project quota limits.

1. In the Cloud Console, go to the Manage resources page.

   Go to the Manage resources page

2. If the project you plan to delete is attached to an organization, select the organization from the Organization list at the top of the page.
3. In the project list, select the project you want to delete and click Delete .
4. In the dialog, type the project ID, and then click Shut down to delete the project.

Alternatively, you can delete the resources used in this tutorial:

1. Delete the Cloud Composer environment.
2. Delete the Cloud Storage bucket for the Cloud Composer environment. Deleting the Cloud Composer environment does not delete its bucket.
3. Delete the Pub/Sub topics for Cloud Composer (composer-agent and composer-backend).

What's next

• Work through the tutorials.
• Get an overview of Cloud Composer.
• Try out other Google Cloud features for yourself. Have a look at our tutorials.
• If you're new to Airflow, check out this tutorial on the Airflow website for more information about concepts, objects, and their usage.