Menjalankan tugas jumlah kata Hadoop di cluster Cloud Dataproc

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Tutorial ini menunjukkan cara menggunakan Cloud Composer untuk membuat Apache Airflow DAG (Directed Acyclic Graph) yang menjalankan tugas jumlah kata Apache Hadoop di Dataproc .

Tujuan

  1. Akses lingkungan Cloud Composer Anda dan gunakan UI Airflow.
  2. Membuat dan melihat variabel lingkungan Airflow.
  3. Buat dan jalankan DAG yang mencakup tugas berikut:
    1. Membuat Dataproc .
    2. Menjalankan Apache Hadoop yang hanya dapat membantu penghitungan kata di cluster.
    3. Menghasilkan hasil jumlah kata ke Penyimpanan Cloud direktori VM dengan bucket.
    4. Menghapus cluster.

Biaya

Dalam dokumen ini, Anda menggunakan komponen Google Cloud yang dapat ditagih berikut:

  • Cloud Composer
  • Dataproc
  • Cloud Storage

Untuk membuat perkiraan biaya berdasarkan proyeksi penggunaan Anda, gunakan kalkulator harga. Pengguna baru Google Cloud mungkin memenuhi syarat untuk mendapatkan uji coba gratis.

Sebelum memulai

  • Pastikan API berikut diaktifkan di project Anda:

    Konsol

    Aktifkan API Dataproc, Cloud Storage.

    Mengaktifkan API

    gcloud

    Aktifkan API Dataproc, Cloud Storage: 

    gcloud services enable dataproc.googleapis.com storage-component.googleapis.com

  • Dalam proyek Anda, membuat bucket Cloud Storage dari semua kelas dan region penyimpanan untuk menyimpan hasil Hadoop jumlah kata.

  • Perhatikan jalur bucket yang Anda buat, misalnya gs://example-bucket. Anda akan menentukan variabel Airflow untuk jalur ini dan menggunakan variabel tersebut dalam contoh DAG nanti dalam tutorial ini.

  • Membuat lingkungan Cloud Composer dengan default parameter. Tunggu hingga pembuatan lingkungan selesai. Setelah selesai, tanda centang hijau ditampilkan di sebelah kiri nama lingkungan.

  • Perhatikan region tempat Anda membuat lingkungan, misalnya us-central. Anda akan menentukan variabel Airflow untuk region ini dan menggunakannya dalam contoh DAG untuk menjalankan cluster Dataproc di region yang sama.

Menyetel variable Airflow

Menetapkan variabel Airflow untuk digunakan nanti dalam contoh DAG. Sebagai contoh, Anda dapat menetapkan variabel Airflow di UI Airflow.

Variabel Airflow Nilai
gcp_project Project ID project yang Anda gunakan untuk tutorial ini, seperti example-project.
gcs_bucket Bucket Cloud Storage URI yang Anda buat untuk tutorial ini, misalnya gs://example-bucket.
gce_region Wilayah tempat Anda membuat lingkungan, seperti us-central1. Ini adalah region tempat cluster Dataproc Anda akan dibuat.

Lihat contoh alur kerja

DAG Airflow adalah kumpulan tugas terorganisir yang ingin Anda jadwalkan lalu jalankan. DAG ditentukan dalam file Python standar. Kode yang ditunjukkan di hadoop_tutorial.py adalah kode alur kerja.

"""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/docs/apache-airflow/stable/concepts/variables.html
* gcp_project - Google Cloud Project to use for the Cloud Dataproc cluster.
* gce_region - Google Compute Engine region 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.providers.google.cloud.operators import dataproc
from airflow.utils import trigger_rule

# Output file for Cloud Dataproc job.
# If you are running Airflow in more than one time zone
# see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
# for best practices
output_file = (
    os.path.join(
        "{{ var.value.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.
input_file = "gs://pub/shakespeare/rose.txt"
wordcount_args = ["wordcount", input_file, output_file]

HADOOP_JOB = {
    "reference": {"project_id": "{{ var.value.gcp_project }}"},
    "placement": {"cluster_name": "composer-hadoop-tutorial-cluster-{{ ds_nodash }}"},
    "hadoop_job": {
        "main_jar_file_uri": WORDCOUNT_JAR,
        "args": wordcount_args,
    },
}

CLUSTER_CONFIG = {
    "master_config": {"num_instances": 1, "machine_type_uri": "n1-standard-2"},
    "worker_config": {"num_instances": 2, "machine_type_uri": "n1-standard-2"},
}

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": "{{ var.value.gcp_project }}",
    "region": "{{ var.value.gce_region }}",
}


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.DataprocCreateClusterOperator(
        task_id="create_dataproc_cluster",
        # Give the cluster a unique name by appending the date scheduled.
        # See https://airflow.apache.org/docs/apache-airflow/stable/macros-ref.html
        cluster_name="composer-hadoop-tutorial-cluster-{{ ds_nodash }}",
        cluster_config=CLUSTER_CONFIG,
        region="{{ var.value.gce_region }}",
    )

    # Run the Hadoop wordcount example installed on the Cloud Dataproc cluster
    # master node.
    run_dataproc_hadoop = dataproc.DataprocSubmitJobOperator(
        task_id="run_dataproc_hadoop", job=HADOOP_JOB
    )

    # Delete Cloud Dataproc cluster.
    delete_dataproc_cluster = dataproc.DataprocDeleteClusterOperator(
        task_id="delete_dataproc_cluster",
        cluster_name="composer-hadoop-tutorial-cluster-{{ ds_nodash }}",
        region="{{ var.value.gce_region }}",
        # 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

Operator

Untuk mengorkestrasi tiga tugas dalam contoh alur kerja, DAG mengimpor tiga operator Airflow berikut:

  • DataprocClusterCreateOperator: Membuat cluster Dataproc.

  • DataProcHadoopOperator: Mengirim tugas jumlah kata Hadoop dan menulis hasilnya ke bucket Cloud Storage.

  • DataprocClusterDeleteOperator: Menghapus cluster untuk menghindari timbulnya cluster biaya Compute Engine yang sedang berjalan.

Dependensi

Anda mengatur tugas yang ingin Anda jalankan dengan cara yang mencerminkan hubungan dan dependensi. Tugas dalam DAG ini berjalan secara berurutan.

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

Penjadwalan

Nama DAG adalah composer_hadoop_tutorial, dan DAG masing-masing berjalan satu kali mereka. Karena start_date yang diteruskan ke default_dag_args adalah disetel ke yesterday, Cloud Composer menjadwalkan alur kerja agar segera dimulai setelah DAG diupload ke bucket lingkungan.

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:

Mengupload DAG ke bucket lingkungan

Cloud Composer menyimpan DAG di folder /dags di bucket lingkungan.

Untuk mengupload DAG:

  1. Di komputer lokal Anda, simpan hadoop_tutorial.py.

  2. Di Konsol Google Cloud, buka halaman Environments.

    Buka Lingkungan

  3. Dalam daftar lingkungan, di kolom DAGs folder untuk lingkungan, klik link DAGs.

  4. Klik Upload file.

  5. Pilih hadoop_tutorial.py di komputer lokal Anda, lalu klik Open.

Cloud Composer menambahkan DAG ke Airflow dan menjadwalkan DAG secara otomatis. Perubahan DAG akan terjadi dalam waktu 3-5 menit.

Mempelajari operasi DAG

Melihat status tugas

Saat mengupload file DAG ke folder dags/ di Cloud Storage, Cloud Composer akan mengurai file. Ketika berhasil diselesaikan, nama alur kerja muncul di listingan DAG, dan alur kerja tersebut diantrekan untuk dijalankan segera.

  1. Untuk melihat status tugas, buka antarmuka web Airflow, lalu klik DAG di toolbar.

  2. Untuk membuka halaman detail DAG, klik composer_hadoop_tutorial. Ini berisi representasi grafis dari tugas alur kerja dan dependensi.

  3. Untuk melihat status setiap tugas, klik Graph View, lalu arahkan mouse ke grafik untuk tiap tugas.

Buat antrean alur kerja lagi

Untuk menjalankan kembali alur kerja dari Graph View:

  1. Di Tampilan Grafik UI Airflow, klik grafis create_dataproc_cluster.
  2. Untuk mereset ketiga tugas tersebut, klik Hapus, lalu klik Oke untuk mengonfirmasi.
  3. Klik create_dataproc_cluster lagi di Graph View.
  4. Untuk mengantrekan alur kerja lagi, klik Run.

Lihat hasil tugas

Anda juga dapat memeriksa status dan hasil composer_hadoop_tutorial alur kerja Anda dengan membuka halaman konsol Google Cloud berikut:

  • Cluster Dataproc: untuk memantau pembuatan cluster dan penghapusan. Perlu diperhatikan bahwa cluster yang dibuat oleh alur kerja bersifat sementara: cluster tersebut hanya ada selama durasi alur kerja dan dihapus sebagai bagian dari dari tugas alur kerja terakhir.

    Buka Cluster Dataproc

  • Tugas Dataproc: untuk melihat atau memantau Apache Hadoop jumlah kata. Klik ID Pekerjaan untuk melihat output log tugas.

    Buka Tugas Dataproc

  • Browser Cloud Storage: untuk melihat hasil jumlah kata di folder wordcount di bucket Cloud Storage yang Anda buat untuk tutorial ini.

    Buka Browser Cloud Storage

Pembersihan

Hapus resource yang digunakan dalam tutorial ini:

  1. Menghapus lingkungan Cloud Composer, termasuk menghapus bucket lingkungan secara manual.

  2. Hapus bucket Cloud Storage yang menyimpan hasil tugas jumlah kata Hadoop.