GSP261

개요

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

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

목표

이 실습에서는 다음 작업을 실행하는 방법을 알아봅니다.

• Cloud 콘솔을 사용하여 Cloud Composer 환경 만들기
• Airflow 웹 인터페이스에서 방향성 비순환 그래프(DAG) 보기 및 실행
• 스토리지에 있는 워드카운트 작업의 결과 보기

설정 및 요건

실습 시작 버튼을 클릭하기 전에

다음 안내를 확인하세요. 실습에는 시간 제한이 있으며 일시중지할 수 없습니다. 실습 시작을 클릭하면 타이머가 시작됩니다. 이 타이머에는 Google Cloud 리소스를 사용할 수 있는 시간이 얼마나 남았는지 표시됩니다.

실무형 실습을 통해 시뮬레이션이나 데모 환경이 아닌 실제 클라우드 환경에서 직접 실습 활동을 진행할 수 있습니다. 실습 시간 동안 Google Cloud에 로그인하고 액세스하는 데 사용할 수 있는 새로운 임시 사용자 인증 정보가 제공됩니다.

이 실습을 완료하려면 다음을 준비해야 합니다.

• 표준 인터넷 브라우저 액세스 권한(Chrome 브라우저 권장)

참고: 이 실습을 실행하려면 시크릿 모드 또는 시크릿 브라우저 창을 사용하세요. 개인 계정과 학생 계정 간의 충돌로 개인 계정에 추가 요금이 발생하는 일을 방지해 줍니다.

• 실습을 완료하기에 충분한 시간---실습을 시작하고 나면 일시중지할 수 없습니다.

참고: 계정에 추가 요금이 발생하지 않도록 하려면 개인용 Google Cloud 계정이나 프로젝트가 이미 있어도 이 실습에서는 사용하지 마세요.

실습을 시작하고 Google Cloud 콘솔에 로그인하는 방법

1. 실습 시작 버튼을 클릭합니다. 실습 비용을 결제해야 하는 경우 결제 수단을 선택할 수 있는 팝업이 열립니다. 왼쪽에는 다음과 같은 항목이 포함된 실습 세부정보 패널이 있습니다.

   • Google 콘솔 열기 버튼
   • 남은 시간
   • 이 실습에 사용해야 하는 임시 사용자 인증 정보
   • 필요한 경우 실습 진행을 위한 기타 정보

2. Google 콘솔 열기를 클릭합니다. 실습에서 리소스가 가동된 후 로그인 페이지가 표시된 다른 탭이 열립니다.

   팁: 두 개의 탭을 각각 별도의 창으로 나란히 정렬하세요.

   참고: 계정 선택 대화상자가 표시되면 다른 계정 사용을 클릭합니다.

3. 필요한 경우 실습 세부정보 패널에서 사용자 이름을 복사하여 로그인 대화상자에 붙여넣습니다. 다음을 클릭합니다.

4. 실습 세부정보 패널에서 비밀번호를 복사하여 시작 대화상자에 붙여넣습니다. 다음을 클릭합니다.

   중요: 왼쪽 패널에 표시된 사용자 인증 정보를 사용해야 합니다. Google Cloud Skills Boost 사용자 인증 정보를 사용하지 마세요. 참고: 이 실습에 자신의 Google Cloud 계정을 사용하면 추가 요금이 발생할 수 있습니다.

5. 이후에 표시되는 페이지를 클릭하여 넘깁니다.

   • 이용약관에 동의합니다.
   • 임시 계정이므로 복구 옵션이나 2단계 인증을 추가하지 않습니다.
   • 무료 평가판을 신청하지 않습니다.

잠시 후 Cloud 콘솔이 이 탭에서 열립니다.

참고: 왼쪽 상단에 있는 탐색 메뉴를 클릭하면 Google Cloud 제품 및 서비스 목록이 있는 메뉴를 볼 수 있습니다.

Cloud Shell 활성화

Cloud Shell은 다양한 개발 도구가 탑재된 가상 머신으로, 5GB의 영구 홈 디렉터리를 제공하며 Google Cloud에서 실행됩니다. Cloud Shell을 사용하면 명령줄을 통해 Google Cloud 리소스에 액세스할 수 있습니다.

1. Google Cloud 콘솔 상단에서 Cloud Shell 활성화 를 클릭합니다.

연결되면 사용자 인증이 이미 처리된 것이며 프로젝트가 PROJECT_ID로 설정됩니다. 출력에 이 세션의 PROJECT_ID를 선언하는 줄이 포함됩니다.

Your Cloud Platform project in this session is set to YOUR_PROJECT_ID

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

2. (선택사항) 다음 명령어를 사용하여 활성 계정 이름 목록을 표시할 수 있습니다.

gcloud auth list

3. 승인을 클릭합니다.

4. 다음과 비슷한 결과가 출력됩니다.

출력:

ACTIVE: * ACCOUNT: student-01-xxxxxxxxxxxx@qwiklabs.net To set the active account, run: $ gcloud config set account `ACCOUNT`

5. (선택사항) 다음 명령어를 사용하여 프로젝트 ID 목록을 표시할 수 있습니다.

gcloud config list project

출력:

[core] project = <project_ID>

출력 예시:

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

작업 1. Cloud Composer 환경 만들기

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

1. 탐색 메뉴() > Composer로 이동합니다.

2. 환경 만들기를 클릭하고 드롭다운에서 Composer 1을 선택합니다.

3. 환경을 다음과 같이 설정합니다.

   • 이름: highcpu

   • 위치:

   • 이미지 버전: 최신 버전 선택

   • 영역:

   • 머신 유형: e2-standard-2

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

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

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

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

Cloud Storage 버킷 만들기

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

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

2. 버킷 이름 을 지정하고 리전에서 버킷을 만든 다음 만들기를 클릭합니다.

공개 액세스가 차단됨 메시지가 표시되면 확인을 클릭합니다.

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

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

Cloud Storage 버킷을 만듭니다.

작업 2. Airflow 및 핵심 개념

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

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

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

핵심 개념

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

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

연산자

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

태스크

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

태스크 인스턴스

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

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

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

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

1. Cloud 콘솔의 오른쪽 상단에 있는 Cloud Shell 활성화 아이콘을 클릭하여 새 Cloud Shell 창을 엽니다.

2. Cloud shell에서 nano(코드 편집기)를 사용하여 hadoop_tutorial.py 파일을 만듭니다.

nano hadoop_tutorial.py

3. 다음은 DAG라고도 하는 hadoop_tutorial.py 워크플로의 코드입니다. 다음 코드를 hadoop_tutorial.py 파일에 붙여넣습니다.

# [START composer_hadoop_tutorial] """Cloud Dataproc 클러스터를 만들고, Hadoop 워드카운트 예를 실행하고, 클러스터를 삭제하는 Airflow DAG의 예입니다. 이 DAG는 3가지 Airflow 변수에 의존합니다. https://airflow.apache.org/concepts.html#variables * gcp_project - Cloud Dataproc 클러스터에 사용할 Google Cloud 프로젝트입니다. * gce_zone - Cloud Dataproc 클러스터를 생성해야 하는 Google Compute Engine 영역입니다. * gcs_bucket - Hadoop 작업 결과에 사용할 Google Cloud Storage 버킷입니다. 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. 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') } # [START composer_hadoop_schedule] 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: # [END composer_hadoop_schedule] # 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, region='{{{project_0.default_region|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 = dataproc_operator.DataProcHadoopOperator( task_id='run_dataproc_hadoop', region='{{{project_0.default_region|place_holder_text}}}', 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', region='{{{project_0.default_region|place_holder_text}}}', 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) # [START composer_hadoop_steps] # Define DAG dependencies. create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster # [END composer_hadoop_steps] # [END composer_hadoop_tutorial]

4. Ctrl+O, Enter, Ctrl+X를 순서대로 눌러 파일을 저장하고 nano를 종료합니다.

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

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

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

# DAG 종속 항목을 정의합니다. create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

DAG 이름은 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 업로드 직후 워크플로가 시작되도록 예약합니다.

작업 4. 환경 정보 보기

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

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

환경 세부정보에는 Airflow 웹 인터페이스 URL, Kubernetes Engine 클러스터 ID, 버킷에 저장된 DAG 폴더에 대한 링크와 같은 정보가 표시됩니다.

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

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

Cloud Composer 환경을 만듭니다.

작업 5. Airflow UI 사용

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

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

작업 6. Airflow 변수 설정

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

1. Airflow 메뉴 바에서 관리 > 변수를 선택한 다음, +(새 레코드 추가)를 클릭합니다.

2. gcp_project, gcs_bucket, gce_zone과 같은 Airflow 변수를 만듭니다.

키	값	세부정보
gcp_project		이 빠른 시작에 사용 중인 Google Cloud 프로젝트입니다.
gcs_bucket	gs://	이전에 만든 Cloud Storage 버킷의 이름입니다. 변경하지 않는 한 프로젝트 ID와 동일합니다. 이 버킷은 Dataproc의 Hadoop 작업 출력을 저장합니다.
gce_zone		Cloud Dataproc 클러스터가 생성될 Compute Engine 영역입니다. 다른 영역을 선택하려면 사용 가능한 리전 및 영역을 참조하세요.

참고: 레코드를 입력한 후 저장을 클릭한 다음 다시 +를 클릭하여 새 레코드를 입력합니다.

완료되면 변수 테이블이 다음과 같이 표시됩니다.

작업 7. Cloud Storage에 DAG 업로드

DAG를 업로드하려면 환경을 만들 때 자동으로 생성된 Cloud Storage 버킷에 hadoop_tutorial.py 파일의 사본을 업로드합니다.

1. Composer > 환경으로 이동하여 확인할 수 있습니다.

2. 이전에 생성한 환경을 클릭하면 생성한 환경에 대한 세부정보로 이동합니다.

3. 환경 구성을 클릭합니다.

4. DAGs folder를 찾아 다음 명령어에서 <DAGs_folder_path>를 대체할 값을 복사하여 Cloud Shell에서 실행합니다.

gsutil cp hadoop_tutorial.py <DAGs_folder_path>

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

Airflow 웹 인터페이스에서 태스크 상태를 볼 수 있습니다.

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

Cloud Storage에 DAG를 업로드 중입니다.

DAG 실행 탐색

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

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

1. Airflow에서 composer_hadoop_tutorial을 클릭하여 DAG 세부정보 페이지를 엽니다. 이 페이지에서는 워크플로 태스크와 종속 항목이 그래픽으로 표시됩니다.

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

3. 필요한 경우 자동 새로고침을 사용 설정하거나 새로고침 아이콘을 클릭하여 최신 정보를 확인하세요. 프로세스의 테두리 색상은 프로세스 상태에 따라 바뀝니다.

create_dataproc_cluster 테두리의 색상이 변경되었고 상태가 '실행 중'이 아닌 경우 그래프 뷰에서 워크플로를 다시 실행합니다. 상태가 '실행 중'이면 아래 3가지 단계를 수행할 필요가 없습니다.

1. create_dataproc_cluster 그래픽을 클릭합니다.
2. 지우기를 클릭하여 태스크 3개를 재설정합니다.
3. 확인을 클릭하여 확인합니다.

create_dataproc_cluster 테두리의 색상이 변경되었으며 상태가 '실행 중'입니다.

Cloud 콘솔에서 프로세스를 모니터링할 수도 있습니다.

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

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

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

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

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

DAG 실행 탐색하기

학습한 내용 테스트

퀴즈를 통해 Google Cloud Platform에 대해 얼마나 알고 있는지 테스트해 보세요.

Cloud Composer 환경 삭제

1. Composer의 Environments 페이지로 돌아갑니다.
2. Composer 환경 옆에 있는 확인란을 선택합니다.
3. DELETE를 클릭합니다.
4. DELETE를 다시 클릭하여 팝업을 확인합니다.

수고하셨습니다

수고하셨습니다. 이 실습에서는 Cloud Composer 환경을 만들고, DAG를 Cloud Storage에 업로드하고, Cloud Dataproc 클러스터를 생성하는 워크플로를 실행하고, Hadoop 워드카운트 작업을 실행하며, 클러스터를 삭제했습니다. Airflow 및 핵심 개념에 대해 알아보고 Airflow 웹 인터페이스도 살펴봤습니다. 이제 Cloud Composer를 사용해 자체 워크플로를 만들고 관리할 수 있게 되었습니다.

다음 단계

이 실습은 Qwik Start 실습 시리즈 중 하나이며 Google Cloud에서 제공하는 다양한 기능을 간략하게 소개하고자 마련되었습니다. 실습 카탈로그에서 'Qwik Start'를 검색하여 다음으로 참여할 실습을 찾아보세요.

• 변수 값을 보려면 get 인수와 함께 Airflow CLI 하위 명령어 변수를 실행하거나 Airflow 웹 인터페이스를 사용합니다.
• Airflow 웹 인터페이스에 대한 자세한 내용은 웹 인터페이스 액세스를 참조하세요.

Google Cloud 교육 및 자격증

Google Cloud 기술을 최대한 활용하는 데 도움이 됩니다. Google 강의에는 빠른 습득과 지속적인 학습을 지원하는 기술적인 지식과 권장사항이 포함되어 있습니다. 기초에서 고급까지 수준별 학습을 제공하며 바쁜 일정에 알맞은 주문형, 실시간, 가상 옵션이 포함되어 있습니다. 인증은 Google Cloud 기술에 대한 역량과 전문성을 검증하고 입증하는 데 도움이 됩니다.

설명서 최종 업데이트: 2023년 12월 18일

실습 최종 테스트: 2023년 12월 18일

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