개요

워크플로는 데이터 분석의 일반적인 주제입니다. 워크플로에는 데이터를 수집, 변환, 분석하여 의미 있는 정보를 파악하는 작업이 포함됩니다. Google Cloud의 워크플로 호스팅 도구는 널리 사용되는 오픈소스 워크플로 도구 Apache Airflow의 호스팅된 버전인 Cloud Composer입니다.

이번 실습에서는 Google Cloud 콘솔을 사용하여 Cloud Composer 환경을 설정합니다. 그런 다음 Cloud Composer를 사용하여 데이터 파일의 유무를 확인하고, Cloud Dataproc 클러스터를 만든 다음, Cloud Dataproc 클러스터에서 Apache Hadoop 워드카운트 작업을 실행한 후에 Cloud Dataproc 클러스터를 삭제하는 간단한 워크플로를 진행합니다.

실습할 내용

• Google Cloud 콘솔을 사용하여 Cloud Composer 환경 만들기

• Airflow 웹 인터페이스에서 방향성 비순환 그래프(DAG) 보기 및 실행

• 스토리지에 있는 워드카운트 작업의 결과 보기

설정 및 요건

실습 설정

각 실습에서는 정해진 기간 동안 새 Google Cloud 프로젝트와 리소스 집합이 무료로 제공됩니다.

1. 시크릿 창을 사용하여 Qwiklabs에 로그인합니다.

2. 실습 사용 가능 시간(예: 1:15:00)을 참고하여 해당 시간 내에 완료합니다.
   일시중지 기능은 없습니다. 필요한 경우 다시 시작할 수 있지만 처음부터 시작해야 합니다.

3. 준비가 되면 실습 시작을 클릭합니다.

4. 실습 사용자 인증 정보(사용자 이름 및 비밀번호)를 기록해 두세요. Google Cloud Console에 로그인합니다.

5. Google Console 열기를 클릭합니다.

6. 다른 계정 사용을 클릭한 다음, 안내 메시지에 이 실습에 대한 사용자 인증 정보를 복사하여 붙여넣습니다.
   다른 사용자 인증 정보를 사용하는 경우 오류가 발생하거나 요금이 부과됩니다.

7. 약관에 동의하고 리소스 복구 페이지를 건너뜁니다.

Google Cloud Shell 활성화하기

Google Cloud Shell은 다양한 개발 도구가 탑재된 가상 머신으로, 5GB의 영구 홈 디렉터리를 제공하며 Google Cloud에서 실행됩니다.

Google Cloud Shell을 사용하면 명령줄을 통해 Google Cloud 리소스에 액세스할 수 있습니다.

1. Cloud 콘솔의 오른쪽 상단 툴바에서 'Cloud Shell 열기' 버튼을 클릭합니다.

   

2. 계속을 클릭합니다.

환경을 프로비저닝하고 연결하는 데 몇 분 정도 소요됩니다. 연결되면 사용자가 미리 인증되어 프로젝트가 PROJECT_ID로 설정됩니다. 예:

gcloud는 Google Cloud의 명령줄 도구입니다. Cloud Shell에 사전 설치되어 있으며 명령줄 자동 완성을 지원합니다.

• 다음 명령어를 사용하여 사용 중인 계정 이름을 나열할 수 있습니다.

gcloud auth list

출력:

Credentialed accounts: - @.com (active)

출력 예시:

Credentialed accounts: - google1623327_student@qwiklabs.net

• 다음 명령어를 사용하여 프로젝트 ID를 나열할 수 있습니다.

gcloud config list project

출력:

[core] project =

출력 예시:

[core] project = qwiklabs-gcp-44776a13dea667a6 참고: gcloud 전체 문서는 gcloud CLI 개요 가이드를 참조하세요.

프로젝트 권한 확인

Google Cloud에서 작업을 시작하기 전에 프로젝트가 Identity and Access Management(IAM) 내에서 올바른 권한을 보유하고 있는지 확인해야 합니다.

1. Google Cloud 콘솔의 탐색 메뉴()에서 IAM 및 관리자 > IAM을 선택합니다.

2. 기본 컴퓨팅 서비스 계정 {project-number}-compute@developer.gserviceaccount.com이 있고 editor 역할이 할당되어 있는지 확인하세요. 계정 프리픽스는 프로젝트 번호이며, 이 번호는 탐색 메뉴 > Cloud 개요 > 대시보드에서 확인할 수 있습니다.

참고: 계정이 IAM에 없거나 editor 역할이 없는 경우 다음 단계에 따라 필요한 역할을 할당합니다.

1. Google Cloud 콘솔의 탐색 메뉴에서 Cloud 개요 > 대시보드를 클릭합니다.
2. 프로젝트 번호(예: 729328892908)를 복사합니다.
3. 탐색 메뉴에서 IAM 및 관리자 > IAM을 선택합니다.
4. 역할 테이블 상단에서 주 구성원별로 보기 아래에 있는 액세스 권한 부여를 클릭합니다.
5. 새 주 구성원 필드에 다음을 입력합니다.

{project-number}-compute@developer.gserviceaccount.com

6. {project-number}는 프로젝트 번호로 바꿉니다.
7. 역할 필드에서 프로젝트(또는 기본) > 편집자를 선택합니다.
8. 저장을 클릭합니다.

작업 1. 환경 작업

1. Google Cloud 콘솔 제목 표시줄에서 Cloud Shell 활성화를 클릭합니다. 메시지가 표시되면 계속을 클릭합니다.

2. 다음 명령어를 실행하여 Compute 개발자 서비스 계정에 Composer 작업자 역할을 할당합니다.

export PROJECT_ID=$(gcloud config get-value project) export PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format="value(projectNumber)") gcloud projects add-iam-policy-binding {{{project_0.project_id}}} \ --member=serviceAccount:$PROJECT_NUMBER-compute@developer.gserviceaccount.com \ --role=roles/composer.worker

3. 다음 명령어를 실행하여 필요한 API가 프로젝트에 확실하게 사용 설정되었는지 확인합니다.

gcloud services disable composer.googleapis.com gcloud services disable artifactregistry.googleapis.com gcloud services disable container.googleapis.com gcloud services enable artifactregistry.googleapis.com gcloud services enable container.googleapis.com gcloud services enable composer.googleapis.com

작업 2. Cloud Composer 환경 만들기

이 섹션에서는 Cloud Composer 환경을 만듭니다.

참고: 계속 진행하기 전에 이전 작업을 수행할 때 필수 API가 사용 설정되었는지 확인합니다. Cloud Composer 환경 만들기를 성공적으로 진행하려면 필수 API가 사용 설정되어 있어야 하므로, 사용 중지되어 있다면 해당 작업을 다시 수행하세요.

1. Google Cloud 콘솔 제목 표시줄의 검색창에 Composer를 입력한 후 제품 및 페이지 섹션에서 Composer를 클릭합니다.

2. 환경 만들기를 클릭하고 Composer 3을 선택합니다. 환경을 다음과 같이 설정합니다.

속성	값
이름	highcpu
위치
이미지 버전	composer-3-airflow-n.n.n-build.n(참고: 사용 가능한 이미지 중 숫자가 가장 높은 최신 버전 선택)
서비스 계정	xxxxx-compute@developer.gserviceaccount.com
복원력 모드	표준 복원력
Airflow 데이터베이스 영역

3. 환경 리소스에서 소형을 선택합니다.

나머지 설정은 모두 기본값으로 둡니다.

4. 만들기를 클릭합니다.

콘솔의 '환경' 페이지에서 환경 이름 왼쪽에 녹색 체크표시가 나타나면 환경 생성 프로세스가 완료된 것입니다.

환경에서 설정 프로세스를 완료하는 데 15~30분이 걸릴 수 있습니다. 환경이 가동되는 동안 실습을 계속 진행합니다.

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다.

Cloud Composer 환경 만들기

Cloud Storage 버킷 만들기

프로젝트에 Cloud Storage 버킷을 만듭니다. 이 버킷은 Dataproc의 Hadoop 작업에 대한 출력으로 사용됩니다.

1. 탐색 메뉴 > Cloud Storage > 버킷으로 이동한 다음 + 만들기를 클릭합니다.

2. 프로젝트 ID()와 같이 버킷에 범용 고유 이름을 지정한 다음 만들기를 클릭합니다. 공개 액세스가 차단됨이라는 메시지가 표시되면 확인을 클릭합니다.

나중에 실습에서 Cloud Storage 버킷을 Airflow 변수로 사용하게 되므로 버킷 이름을 기억해 두세요.

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다.

Cloud Storage 버킷 만들기

작업 3. Airflow 및 핵심 개념

Composer 환경이 생성되는 것을 기다리는 동안 Airflow에서 사용되는 몇 가지 용어를 검토합니다.

Airflow는 프로그래매틱 방식으로 워크플로를 작성, 예약, 모니터링하는 플랫폼입니다.

Airflow를 사용하여 태스크의 방향성 비순환 그래프(DAG)로 워크플로를 작성합니다. Airflow 스케줄러는 지정된 종속 항목을 준수하면서 작업자 배열에서 태스크를 실행합니다.

핵심 개념

방향성 비순환 그래프(DAG)

방향성 비순환 그래프는 실행하려는 모든 태스크의 모음으로, 관계와 종속 항목을 반영하는 방식으로 구성됩니다.

연산자

단일 태스크에 대한 설명으로 보통 원자적입니다. 예를 들어, BashOperator는 bash 명령어를 실행하는 데 사용됩니다.

태스크

연산자의 매개변수화된 인스턴스로, DAG의 노드입니다.

태스크 인스턴스

특정한 태스크 실행으로 DAG, 태스크, 특정 시점으로 구분합니다. 상태 표시로 실행 중, 성공, 실패, 건너뜀 등이 있습니다.

개념에 관한 자세한 내용은 개념 문서에서 확인할 수 있습니다.

작업 4. 워크플로 정의하기

이제 사용할 워크플로에 대해 알아보겠습니다. Cloud Composer 워크플로는 방향성 비순환 그래프(DAG)로 구성됩니다. DAG는 Airflow의 DAG_FOLDER에 있는 표준 Python 파일에 정의되어 있습니다. Airflow는 각 파일에서 코드를 실행하여 DAG 객체를 동적으로 빌드합니다. DAG를 원하는 만큼 추가할 수 있으며 각 DAG는 임의의 수의 태스크를 설명합니다. 일반적으로 각 DAG는 하나의 논리 워크플로에 대응해야 합니다.

아래 코드는 hadoop_tutorial.py 워크플로 코드로 DAG라고도 합니다.

"""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. * gce_region - Google Compute Engine region where Cloud Dataproc cluster should be created. * gcs_bucket - Google Cloud Storage bucket to used as output for the Hadoop jobs from Dataproc. 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.dataproc import ( DataprocCreateClusterOperator, DataprocDeleteClusterOperator, DataprocSubmitJobOperator, ) from airflow.utils.trigger_rule import TriggerRule # Define the DAG's schedule using a cron expression for daily runs DAILY_SCHEDULE = '0 0 * * *' # Runs at midnight every day # 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( dag_id='composer_hadoop_tutorial', # It's good practice to explicitly use 'dag_id' schedule_interval=DAILY_SCHEDULE, # Using the new cron expression default_args=default_dag_args) as dag: # Create a Cloud Dataproc cluster. create_dataproc_cluster = DataprocCreateClusterOperator( 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, region=models.Variable.get('gce_region'), zone=models.Variable.get('gce_zone'), image_version='2.0', master_machine_type='e2-standard-2', worker_machine_type='e2-standard-2') # Run the Hadoop wordcount example installed on the Cloud Dataproc cluster # master node. run_dataproc_hadoop = DataprocSubmitJobOperator( task_id='run_dataproc_hadoop', region=models.Variable.get('gce_region'), job={ "placement": {"cluster_name": 'composer-hadoop-tutorial-cluster-{{ ds_nodash }}'}, "hadoop_job": { "main_jar_file_uri": WORDCOUNT_JAR, "args": wordcount_args } } ) # Delete Cloud Dataproc cluster. delete_dataproc_cluster = DataprocDeleteClusterOperator( task_id='delete_dataproc_cluster', project_id=models.Variable.get('gcp_project'), region=models.Variable.get('gce_region'), 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=TriggerRule.ALL_DONE) # Define DAG dependencies. create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

워크플로 태스크 3개를 조정하기 위해 DAG는 다음 연산자를 가져옵니다.

1. DataprocCreateClusterOperator: Cloud Dataproc 클러스터를 만듭니다.
2. DataprocSubmitJobOperator: Hadoop 워드카운트 작업을 제출하고 결과를 Cloud Storage 버킷에 씁니다.
3. DataprocDeleteClusterOperator: Compute Engine 요금이 지속적으로 청구되지 않도록 클러스터를 삭제합니다.

태스크는 순차적으로 실행되며 파일의 이 섹션에서 볼 수 있습니다.

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

DAG 이름은 composer_hadoop_tutorial이며 매일 1회 실행됩니다.

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:

default_dag_args에 전달되는 start_date가 yesterday로 설정되어 있기 때문에 Cloud Composer는 DAG 업로드 직후 워크플로가 시작되도록 예약합니다.

작업 5. 환경 정보 보기

1. Composer로 돌아가서 환경 상태를 확인하세요.

2. 환경이 생성되면 환경 이름(highcpu)을 클릭하여 세부정보를 확인합니다.

환경 구성 탭에는 Airflow 웹 UI URL, GKE 클러스터, 버킷에 저장된 DAG 폴더에 대한 링크와 같은 정보가 표시됩니다.

참고: Cloud Composer는 /dags 폴더의 워크플로만 예약합니다.

작업 6. Airflow UI 사용

콘솔을 사용하여 Airflow 웹 인터페이스에 액세스하려면 다음 안내를 따르세요.

1. 환경 페이지로 돌아갑니다.
2. 환경의 Airflow 웹 서버 열에서 Airflow를 클릭합니다.
3. 실습 사용자 인증 정보를 클릭합니다.
4. Airflow 웹 인터페이스가 새 브라우저 창에서 열립니다.

작업 7. Airflow 변수 설정

Airflow 변수는 환경 변수와 다른 Airflow 관련 개념입니다.

1. Airflow 인터페이스의 메뉴 바에서 관리 > 변수를 선택합니다.

2. + 아이콘을 클릭하여 새 레코드를 추가합니다.

3. gcp_project, gcs_bucket, gce_zone, gce_region과 같은 Airflow 변수를 만듭니다. 변수마다 저장을 클릭합니다.

키	Val	세부정보
gcp_project		이 실습에 사용 중인 Google Cloud Platform 프로젝트입니다.
gcs_bucket	gs://	이 버킷은 Dataproc의 Hadoop 작업 출력을 저장합니다.
gce_zone		Cloud Dataproc 클러스터가 생성될 Compute Engine 영역입니다.
gce_region		Cloud Dataproc 클러스터가 생성될 Compute Engine 리전입니다.

저장을 클릭합니다. 첫 번째 변수를 추가한 후 두 번째 및 세 번째 변수에도 동일한 과정을 반복합니다. 완료되면 변수 테이블이 다음과 같이 표시됩니다.

작업 8. Cloud Storage에 DAG 업로드

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

1. Cloud Shell에서 아래 명령어를 실행하여 환경을 만들 때 자동으로 생성된 Cloud Storage 버킷에 hadoop_tutorial.py 파일 사본을 업로드합니다.

2. 다음 명령어에서 <DAGs_folder_path>를 DAG 폴더의 경로로 바꿉니다.

gcloud storage cp gs://cloud-training/datawarehousing/lab_assets/hadoop_tutorial.py <DAGs_folder_path>

• Composer에서 경로를 확인할 수 있습니다.
• 이전에 생성한 환경을 클릭한 다음 환경 구성 탭을 클릭하여 환경의 자세한 정보를 살펴보세요.
• DAG 폴더를 찾고 경로를 복사합니다.

파일 업로드를 위해 수정된 명령어는 다음 명령어와 유사해야 합니다.

gcloud storage cp gs://cloud-training/datawarehousing/lab_assets/hadoop_tutorial.py gs://{{{project_0.default_region|REGION}}}-highcpu-0682d8c0-bucket/dags

3. 파일이 DAG 디렉터리에 성공적으로 업로드되면 버킷의 dags 폴더를 엽니다. 버킷 세부정보의 객체 탭에서 파일을 찾을 수 있습니다.

DAG 파일이 DAG 폴더에 추가되면 Cloud Composer가 DAG를 Airflow에 추가하고 자동으로 일정을 예약합니다. DAG 변경사항은 3~5분 이내에 적용됩니다.

Airflow 웹 인터페이스에서 composer_hadoop_tutorial DAG의 작업 상태를 확인할 수 있습니다.

참고: '스케줄러가 실행되지 않은 것으로 보입니다'와 같은 인터페이스상의 메시지는 무시해도 됩니다. DAG가 진행됨에 따라 Airflow 웹 인터페이스가 업데이트됩니다.

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다.

Cloud Storage에 DAG 업로드

DAG 실행 탐색

DAG 파일을 Cloud Storage의 dags 폴더에 업로드하면 Cloud Composer가 파일을 파싱합니다. 오류가 발견되지 않으면 워크플로 이름이 DAG 목록에 나타나고 워크플로가 즉시 실행되도록 큐에 추가됩니다.

1. Airflow 웹 인터페이스의 DAG 탭이 표시되었는지 확인합니다. 이 프로세스를 완료하는 데 몇 분 정도 걸립니다. 최신 정보가 표시되도록 브라우저를 새로고침하세요.

2. Airflow에서 composer_hadoop_tutorial을 클릭하여 DAG 세부정보 페이지를 엽니다. 이 페이지에는 워크플로 태스크 및 종속 항목이 다양한 형태로 표시됩니다.

3. 툴바에서 그래프를 클릭합니다. 각 태스크의 그래픽에 마우스를 가져가 상태를 확인합니다. 각 태스크 주위의 테두리도 상태를 나타냅니다(예: 녹색 테두리 = 실행 중, 빨간색 테두리 = 실패).

4. '새로고침' 링크를 클릭하여 최신 정보를 확인하세요. 프로세스의 테두리 색상은 프로세스 상태에 따라 바뀝니다.

참고: Dataproc 클러스터가 이미 있는 경우 `create_dataproc_cluster` 그래픽을 클릭하여 성공 상태가 되도록 워크플로를 다시 실행할 수 있으며, 지우기를 클릭하여 세 가지 태스크를 리셋하고 확인을 클릭하면 됩니다.

5. create_dataproc_cluster의 상태가 '실행 중'으로 변경되면 탐색 메뉴 > Dataproc으로 이동한 후 다음을 클릭합니다.

   • 클러스터 생성 및 삭제를 모니터링할 클러스터. 워크플로에서 생성된 클러스터는 수명이 짧습니다. 즉, 워크플로 지속 시간 동안만 존재하며 마지막 워크플로 태스크의 일부로 삭제됩니다.
   • Apache Hadoop 워드카운트 작업을 모니터링하는 작업. 작업 ID를 클릭하여 작업 로그 출력을 확인합니다.

6. Dataproc이 '실행 중' 상태가 되면 Airflow로 돌아가서 새로고침을 클릭하여 클러스터가 완료되었는지 확인합니다.

run_dataproc_hadoop 프로세스가 완료되면 탐색 메뉴 > Cloud Storage > 버킷으로 이동하고 버킷 이름을 클릭하여 wordcount 폴더의 워드카운트 결과를 확인합니다.

7. DAG에서 모든 단계가 완료되면 각 단계에 진한 녹색 테두리가 표시됩니다. 그리고 이전에 생성되었던 Dataproc 클러스터가 삭제됩니다.

수고하셨습니다

Cloud Composer 워크플로를 성공적으로 실행했습니다.

다음 단계

• 샌프란시스코에서 열린 NEXT 18에서 Cloud Composer가 소개된 것을 확인해 보세요. Cloud Composer를 사용하는 Google Cloud의 유연하고 간편한 데이터 파이프라인(Cloud Next '18)
• 변수 값을 보려면 get 인수와 함께 Airflow CLI 하위 명령어 variables를 실행하거나 Airflow 웹 인터페이스를 사용합니다.
• Airflow 웹 인터페이스에 대한 정보는 웹 인터페이스 액세스를 참조하세요.

실습 종료하기

실습을 완료하면 실습 종료를 클릭합니다. Google Cloud Skills Boost에서 사용된 리소스를 자동으로 삭제하고 계정을 지웁니다.

실습 경험을 평가할 수 있습니다. 해당하는 별표 수를 선택하고 의견을 입력한 후 제출을 클릭합니다.

별점의 의미는 다음과 같습니다.

• 별표 1개 = 매우 불만족
• 별표 2개 = 불만족
• 별표 3개 = 중간
• 별표 4개 = 만족
• 별표 5개 = 매우 만족

의견을 제공하고 싶지 않다면 대화상자를 닫으면 됩니다.

의견이나 제안 또는 수정할 사항이 있다면 지원 탭을 사용하세요.

Copyright 2025 Google LLC All rights reserved. Google 및 Google 로고는 Google LLC의 상표입니다. 기타 모든 회사명 및 제품명은 해당 업체의 상표일 수 있습니다.