Dataproc 클러스터에서 Hadoop 워드카운트 작업 실행

이 가이드에서는 Google Cloud Console을 통해 Cloud Composer를 사용하여 Dataproc 클러스터에서 Apache Hadoop 워드카운트 작업을 실행하는 Apache Airflow DAG(워크플로)를 만드는 방법을 설명합니다.

목표

Cloud Composer 환경에 액세스하고 Airflow 웹 인터페이스를 사용합니다.
Airflow 환경 변수를 만들고 봅니다.
다음 작업이 포함된 DAG를 만들고 실행합니다.
1. Dataproc 클러스터를 만듭니다.
2. 클러스터에서 Apache Hadoop 워드카운트 작업을 실행합니다.
3. 워드카운트 결과를 Cloud Storage 버킷에 출력합니다.
4. 클러스터를 삭제합니다.

비용

이 가이드에서는 비용이 청구될 수 있는 다음과 같은 Google Cloud 구성요소를 사용합니다.

Cloud Composer
Dataproc
Cloud Storage

프로젝트 사용량을 기준으로 예상 비용을 산출하려면 가격 계산기를 사용하세요. Google Cloud를 처음 사용하는 사용자는 무료 체험판을 사용할 수 있습니다.

시스템에서 개발자 환경을 만드는 데 최대 25분이 걸립니다. 이 튜토리얼을 완료하는 데 약 1시간 정도 걸릴 수 있습니다.

시작하기 전에

Google Cloud 계정에 로그인합니다. Google Cloud를 처음 사용하는 경우 계정을 만들고 Google 제품의 실제 성능을 평가해 보세요. 신규 고객에게는 워크로드를 실행, 테스트, 배포하는 데 사용할 수 있는 $300의 무료 크레딧이 제공됩니다.

Google Cloud Console의 프로젝트 선택기 페이지에서 Google Cloud 프로젝트를 선택하거나 만듭니다.

프로젝트 선택기로 이동

Cloud 프로젝트에 결제가 사용 설정되어 있는지 확인합니다. 프로젝트에 결제가 사용 설정되어 있는지 확인하는 방법을 알아보세요.

Cloud Composer, Cloud Dataproc, and Cloud Storage API를 사용 설정합니다.

API 사용 설정

Google Cloud Console의 프로젝트 선택기 페이지에서 Google Cloud 프로젝트를 선택하거나 만듭니다.

프로젝트 선택기로 이동

Cloud 프로젝트에 결제가 사용 설정되어 있는지 확인합니다. 프로젝트에 결제가 사용 설정되어 있는지 확인하는 방법을 알아보세요.

Cloud Composer, Cloud Dataproc, and Cloud Storage API를 사용 설정합니다.

API 사용 설정

프로젝트에서 스토리지 클래스와 리전에 상관없이 Hadoop 워드카운트 작업 결과를 저장할 Cloud Storage 버킷을 만듭니다.
데이터 보존 요구사항 또는 성능상의 이점을 위해 환경을 만들려는 리전과 동일한 리전에 스토리지 버킷을 만듭니다.
만든 버킷의 경로를 기록합니다(예: gs://my-bucket). 이 경로에 Airflow 변수를 정의하여 예시 DAG에서 사용합니다.

환경 만들기

Cloud Console에서 환경 만들기 페이지로 이동합니다.

환경 만들기 페이지 열기
이름 필드에 example-environment를 입력합니다.
위치 드롭다운 목록에서 Cloud Composer 환경의 리전을 선택합니다. 리전 선택에 대한 자세한 내용은 사용 가능한 리전을 참조하세요.
다른 환경 구성 옵션에 제공된 기본값을 사용합니다.
환경을 만들려면 만들기를 클릭합니다.
환경 생성이 완료될 때까지 기다립니다. 작업이 완료되면 환경 이름의 왼쪽에 녹색 체크표시가 표시됩니다.

환경 세부정보 보기

환경 생성이 완료되면 Cloud Composer 및 Python 버전, Airflow 웹 인터페이스의 URL, Google Kubernetes Engine 클러스터 ID와 같은 환경의 배포 정보를 볼 수 있습니다.

배포 정보를 보려면 다음 안내를 따르세요.

Cloud Console에서 환경 페이지로 이동합니다.

환경 페이지 열기
환경 세부정보 페이지를 보려면 example-environment를 클릭합니다.
환경을 만든 영역을 기록합니다(예: us-central-1c). 이 영역에 Airflow 변수를 정의하여 예시 DAG에서 사용합니다.

Airflow 변수 설정

Airflow 변수는 환경 변수와 다른 Airflow 관련 개념입니다. 이 단계에서는 Airflow 웹 인터페이스를 통해 나중에 예시 DAG에서 사용할 Airflow 변수 세 개를 설정합니다.

변수를 설정하려면 다음 안내를 따르세요.

Cloud Console에서 Airflow 웹 인터페이스에 액세스합니다.
1. Cloud Console에서 환경 페이지로 이동합니다.
  
  환경 페이지 열기
2. example-environment의 Airflow 웹 서버 열에서 Airflow 링크를 클릭합니다. Airflow 웹 인터페이스가 새 창에서 열립니다.

Airflow 웹 인터페이스에서 변수를 설정합니다.

툴바에서 관리 > 변수를 클릭합니다.
새 변수를 만들려면 만들기를 클릭합니다.

다음 변수 각각에 키-값 쌍을 입력하고 저장을 클릭합니다. 모든 Airflow 변수가 목록 탭에 표시됩니다.

키	값
gcp_project	이 가이드를 위해 사용 중인 `composer-test`와 같은 Google Cloud Platform 프로젝트의 프로젝트 ID입니다.
gcs_bucket	이 가이드용으로 만든 Cloud Storage 버킷(예: `gs://my-bucket`)
gce_zone	환경에 사용할 Compute Engine 영역(예: `us-central1-c`). Dataproc 클러스터를 만들 영역입니다. 사용 가능한 리전 및 영역을 참조하세요.

Cloud SDK를 설치하거나 Cloud Shell을 사용할 경우 gcloud 명령줄 도구를 사용하여 Airflow CLI 하위 명령어를 실행할 수 있습니다. 다음 예시 gcloud composer 명령어를 variables 하위 명령어와 함께 사용하여 Airflow 변수 gcp_project를 project-id로 설정합니다.

For Airflow 1.10 CLI: gcloud composer environments run environment-name --location us-central1 variables -- --set gcp_project project-id

For Airflow 2.0 CLI: gcloud beta composer environments run environment-name --location us-central1 variables set -- gcp_project project-id

gcloud 명령어 사용에 대한 자세한 내용은 Cloud SDK 사용을 참조하세요.

예시 워크플로 보기

Airflow DAG는 예약 및 실행하려는 태스크가 구성된 모음입니다. DAG는 표준 Python 파일에서 정의됩니다. hadoop_tutorial.py에 나와 있는 코드는 워크플로 코드입니다.

Airflow 2

composer/workflows/hadoop_tutorial.py

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/docs/apache-airflow/stable/concepts/variables.html
* 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.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(
    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.
input_file = 'gs://pub/shakespeare/rose.txt'
wordcount_args = ['wordcount', input_file, output_file]

HADOOP_JOB = {
    "reference": {"project_id": models.Variable.get('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': models.Variable.get('gcp_project'),
    'location': models.Variable.get('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=models.Variable.get('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=models.Variable.get('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

Airflow 1

composer/airflow_1_samples/hadoop_tutorial.py

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/docs/apache-airflow/stable/concepts/variables.html
* 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.
# 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(
    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.
input_file = 'gs://pub/shakespeare/rose.txt'
wordcount_args = ['wordcount', input_file, 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/docs/apache-airflow/stable/macros-ref.html
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        num_workers=2,
        zone=models.Variable.get('gce_zone'),
        master_machine_type='n1-standard-2',
        worker_machine_type='n1-standard-2')

    # 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

연산자

연산자는 워크플로의 단일 태스크용 템플릿입니다. 예시 워크플로에서 태스크 세 개를 조정하기 위해 DAG는 다음 연산자 세 개를 가져옵니다.

DataprocClusterCreateOperator: Dataproc 클러스터를 만듭니다.
DataProcHadoopOperator: Hadoop 워드카운트 작업을 제출하고 결과를 Cloud Storage 버킷에 기록합니다.
DataprocClusterDeleteOperator: Compute Engine 요금이 청구되지 않도록 클러스터를 삭제합니다.

종속 항목

실행할 태스크를 관계 및 종속 항목을 반영하는 방식으로 구성합니다. 이 DAG의 태스크는 순차적으로 실행됩니다. 이 예시에서 관계는 Python bitshift 연산자가 가리키는 방향으로 설정됩니다(>>).

Airflow 2

composer/workflows/hadoop_tutorial.py

GitHub에서 보기

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

Airflow 1

composer/airflow_1_samples/hadoop_tutorial.py

GitHub에서 보기

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

예약

DAG 이름은 composer_hadoop_tutorial이며 매일 1회 실행됩니다. default_dag_args로 전달된 start_date가 yesterday로 설정되어 있으므로 Cloud Composer는 DAG 업로드 직후에 워크플로가 시작되도록 예약합니다.

Airflow 2

composer/workflows/hadoop_tutorial.py

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:

Airflow 1

composer/airflow_1_samples/hadoop_tutorial.py

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:

Cloud Storage에 DAG 업로드

Cloud Composer는 DAG 폴더의 DAG만 예약합니다. DAG 폴더는 Cloud Composer가 환경에 맞게 자동으로 만드는 Cloud Storage 버킷에 위치합니다.

DAG를 업로드하려면 다음 안내를 따르세요.

로컬 머신에서 hadoop_tutorial.py를 저장합니다.
Cloud Console에서 환경 페이지로 이동합니다.

환경 페이지 열기
example-environment의 DAGs 폴더 열에서 DAGs 링크를 클릭합니다. Cloud Storage의 DAG 폴더가 열립니다.
파일 업로드를 클릭합니다.
로컬 머신에서 hadoop_tutorial.py를 선택하고 열기를 클릭합니다.

Cloud Composer는 DAG를 Airflow에 추가하고 DAG를 자동으로 예약합니다. DAG 변경사항은 3~5분 이내에 적용됩니다.

DAG 실행 탐색

태스크 상태 보기

DAG 파일을 Cloud Storage의 dags/ 폴더에 업로드하면 Cloud Composer가 파일을 파싱합니다. 작업이 성공적으로 완료되면 워크플로 이름이 DAG 목록에 나타나고 워크플로가 즉시 실행되도록 큐에 추가됩니다.

태스크 상태를 보려면 Airflow 웹 인터페이스로 이동하고 툴바에서 DAGs를 클릭합니다.
DAG 세부정보 페이지를 열려면 composer_hadoop_tutorial을 클릭합니다. 이 페이지에서는 워크플로 태스크와 종속 항목이 그래픽으로 표시됩니다.
각 태스크의 상태를 확인하려면 그래프 뷰를 클릭한 후 각 태스크의 그래픽에 마우스를 가져갑니다.

워크플로를 다시 큐에 추가

그래프 뷰에서 워크플로를 다시 실행하려면 다음 안내를 따르세요.

Airflow UI 그래프 뷰에서 create_dataproc_cluster 그래픽을 클릭합니다.
태스크 세 개를 재설정하려면 삭제를 클릭한 후 확인을 클릭하여 확인합니다.
그래프 뷰에서 create_dataproc_cluster를 다시 클릭합니다.
워크플로를 다시 큐에 추가하려면 실행을 클릭합니다.

태스크 결과 보기

다음 Cloud Console 페이지로 이동하여 composer_hadoop_tutorial 워크플로의 상태와 결과를 확인할 수도 있습니다.

클러스터 생성 및 삭제를 모니터링할 Dataproc 클러스터. 워크플로에서 생성된 클러스터는 수명이 짧습니다. 즉, 워크플로 지속시간 동안만 존재하며 마지막 워크플로 태스크의 일부로 삭제됩니다.
Apache Hadoop 워드카운트 작업을 보거나 모니터링하기 위한 Dataproc 작업. 작업 ID를 클릭하여 작업 로그 출력을 확인합니다.
이 가이드에서 만든 Cloud Storage 버킷에 있는 wordcount 폴더의 워드카운트 결과를 확인하기 위한 Cloud Storage 브라우저

정리

이 가이드에서 사용된 리소스 비용이 Google Cloud 계정에 청구되지 않도록 하려면 리소스가 포함된 프로젝트를 삭제하거나 프로젝트를 유지하고 개별 리소스를 삭제하세요.

주의: 프로젝트를 삭제하면 다음과 같은 효과가 발생합니다.

프로젝트의 모든 항목이 삭제됩니다. 이 가이드의 기존 프로젝트를 사용한 경우 프로젝트를 삭제하면 프로젝트에서 수행한 다른 작업도 삭제됩니다.
커스텀 프로젝트 ID가 손실됩니다. 이 프로젝트를 만들 때 앞으로 사용할 커스텀 프로젝트 ID를 만들었을 수 있습니다. appspot.com URL과 같이 프로젝트 ID를 사용하는 URL을 보존하려면 전체 프로젝트를 삭제하는 대신 프로젝트 내에서 선택한 리소스만 삭제합니다.

여러 가이드와 빠른 시작을 살펴보려는 경우 프로젝트를 재사용하면 프로젝트 할당량 한도 초과를 방지할 수 있습니다.

Cloud Console에서 리소스 관리 페이지로 이동합니다.
리소스 관리로 이동
삭제하려는 프로젝트가 조직에 연결되어 있으면 이름 열에서 조직 목록을 확장합니다.
프로젝트 목록에서 삭제할 프로젝트를 선택하고 삭제를 클릭합니다.
대화상자에서 프로젝트 ID를 입력한 후 종료를 클릭하여 프로젝트를 삭제합니다.

또는 이 가이드에서 사용된 리소스를 삭제할 수 있습니다.

Cloud Composer 환경을 삭제합니다.
Cloud Composer 환경의 Cloud Storage 버킷을 삭제합니다. Cloud Composer 환경을 삭제하면 버킷이 삭제되지 않습니다.
Cloud Composer의 Pub/Sub 주제를 삭제합니다(composer-agent 및 composer-backend)).

다음 단계

가이드 살펴보기
Cloud Composer 개요 보기
Google Cloud에 대한 참조 아키텍처, 다이어그램, 가이드, 권장사항 살펴보기 Cloud 아키텍처 센터 살펴보기
Airflow를 처음 사용할 경우 개념, 객체, 사용법에 대한 자세한 내용은 Airflow 웹사이트의 이 가이드를 참조하세요.