GSP808

개요

일괄 파이프라인 외에도 Data Fusion을 사용하면 이벤트가 생성될 때 이를 처리할 수 있는 실시간 파이프라인을 만들 수 있습니다. 현재 실시간 파이프라인은 Cloud Dataproc 클러스터에서 Apache Spark 스트리밍을 사용하여 실행됩니다. 이 실습에서는 Data Fusion을 사용하여 스트리밍 파이프라인을 빌드하는 방법을 배웁니다.

Cloud Pub/Sub 주제에서 읽고 이벤트를 처리하며 일부 변환을 실행하고 출력을 BigQuery에 쓰는 파이프라인을 만들게 됩니다.

목표

1. 실시간 파이프라인을 만드는 방법을 알아봅니다.
2. Data Fusion에서 Pub/Sub 소스 플러그인을 구성하는 방법을 알아봅니다.
3. Wrangler를 사용하여 지원되지 않는 연결에 있는 데이터의 변환을 정의하는 방법을 알아봅니다.

설정 및 요건

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

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

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

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

   참고: 실습 시작을 클릭하고 나면 실습에서 필요한 리소스를 프로비저닝하고 Data Fusion 인스턴스를 만드는 데 약 15~20분이 걸립니다. 이 시간 동안 아래의 단계를 읽으면서 실습 목표를 숙지하세요.
   왼쪽 패널에 실습 사용자 인증 정보(사용자 이름 및 비밀번호)가 보이면 인스턴스가 생성된 것이므로 콘솔에 계속 로그인할 수 있습니다.

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

5. Google 콘솔 열기를 클릭합니다.

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

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

참고: 실습을 완료했거나 다시 시작하려고 하는 경우가 아니면 실습 종료를 클릭하지 마세요. 클릭하면 작업이 지워지고 프로젝트가 삭제됩니다.

Google Cloud 콘솔에 로그인

1. 이 실습 세션에 사용하는 브라우저 탭 또는 창을 활용해 연결 세부정보 패널에서 사용자 이름을 복사하고 Google 콘솔 열기 버튼을 클릭합니다.

참고: 계정을 선택하라는 메시지가 표시되면 다른 계정 사용을 클릭합니다.

2. 안내에 따라 사용자 이름과 비밀번호를 붙여넣습니다.
3. 다음을 클릭합니다.
4. 이용약관에 동의합니다.

임시 계정이므로 이 실습이 진행되는 동안만 계정이 유지됩니다.

• 복구 옵션을 추가하지 않습니다.
• 무료 체험판을 신청하지 않습니다.

5. 콘솔이 열리면 왼쪽 상단의 탐색 메뉴()를 클릭해 서비스 목록을 확인합니다.

Cloud Shell 활성화

Cloud Shell은 여러 개발 도구가 포함된 가상 머신입니다. 5GB의 영구적인 홈 디렉터리를 제공하며 Google Cloud에서 실행됩니다. Cloud Shell을 사용하면 명령줄을 통해 Google Cloud 리소스에 액세스할 수 있습니다. gcloud는 Google Cloud의 명령줄 도구입니다. Cloud Shell에 사전 설치되어 있으며 탭 자동 완성을 지원합니다.

1. Google Cloud Console의 탐색창에서 Cloud Shell 활성화()를 클릭합니다.

2. 계속을 클릭합니다.
   환경을 프로비저닝하고 연결하는 데는 몇 분 정도 소요됩니다. 연결되면 사용자 인증도 처리되어 프로젝트가 PROJECT_ID로 설정됩니다. 예를 들면 다음과 같습니다.

샘플 명령어

• 활성 계정 이름을 나열합니다.

gcloud auth list

(출력)

Credentialed accounts: - <myaccount>@<mydomain>.com (active)

(출력 예시)

Credentialed accounts: - google1623327_student@qwiklabs.net

• 프로젝트 ID를 나열합니다.

gcloud config list project

(출력)

[core] project = <project_ID>

(출력 예시)

[core] project = qwiklabs-gcp-44776a13dea667a6

작업 1. 프로젝트 권한

프로젝트 권한 확인

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. 프로젝트 정보 카드에서 프로젝트 번호를 복사합니다.

3. 탐색 메뉴에서 IAM 및 관리자 > IAM을 클릭합니다.

4. IAM 페이지 상단에서 추가를 클릭합니다.

5. 새 주 구성원 필드에 다음을 입력합니다.

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

{project-number}는 프로젝트 번호로 바꿉니다.

6. 역할 선택에서 기본 (또는 프로젝트) > 편집자를 선택합니다.

7. 저장을 클릭합니다.

작업 2. Dataflow API가 사용 설정되어 있는지 확인

필요한 API에 액세스할 수 있도록 Dataflow API에 대한 연결을 다시 시작합니다.

1. Cloud 콘솔의 상단 검색창에 'Dataflow API'를 입력합니다. Dataflow API에 대한 검색 결과를 클릭합니다.

2. 관리를 클릭합니다.

3. API 사용 중지를 클릭합니다.

확인을 요청하는 메시지가 표시되면 사용 중지를 클릭합니다.

4. 사용 설정을 클릭합니다.

작업 3. 데이터 로드

1. 먼저 컴퓨터에 샘플 트윗을 다운로드해야 합니다. 나중에 Wrangler를 사용해 이를 업로드하여 변환 단계를 만들게 됩니다.

또한 Cloud Storage 버킷에 동일한 샘플 트윗 파일을 스테이징해야 합니다. 이 실습의 마지막 부분에서는 버킷에서 Pub/Sub 주제로 데이터를 스트리밍합니다.

2. Cloud Shell에서 다음 명령어를 실행하여 새 버킷을 만듭니다.

export BUCKET=$GOOGLE_CLOUD_PROJECT gsutil mb gs://$BUCKET

생성된 버킷의 이름은 프로젝트 ID와 동일합니다.

3. 아래 명령어를 실행하여 트윗 파일을 버킷에 복사합니다.

gsutil cp gs://cloud-training/OCBL164/pubnub_tweets_2019-06-09-05-50_part-r-00000 gs://$BUCKET

4. 파일이 Cloud Storage 버킷에 복사되었는지 확인합니다.

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다. 데이터 로드

작업 4. Pub/Sub 주제 설정

Pub/Sub를 사용하려면 데이터를 포함할 주제를 만들고 이 주제에 게시된 데이터에 액세스할 수 있는 구독을 만들어야 합니다.

1. Cloud 콘솔의 탐색 메뉴에서 모든 제품 보기를 클릭하고 분석 섹션에서 Pub/Sub를 클릭한 다음 주제를 선택합니다.

2. 주제 만들기를 클릭합니다.

3. 주제에는 고유한 이름을 지정해야 합니다. 이 실습에서는 주제의 이름을 cdf_lab_topic으로 지정한 다음 만들기를 클릭합니다.

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다. Pub/Sub 주제 설정

작업 5. Pub/Sub 구독 추가

계속해서 주제 페이지에서 진행합니다. 이제 주제에 액세스할 수 있도록 구독을 만들어 보겠습니다.

1. 구독 만들기를 클릭합니다.

2. cdf_lab_subscription 등 구독의 이름을 입력하고 '전송 유형'을 가져오기로 설정한 후 만들기를 클릭합니다.

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다. Pub/Sub 구독 추가

작업 6. Cloud Data Fusion 인스턴스에 필요한 권한 추가

1. Google Cloud 콘솔의 탐색 메뉴에서 모든 제품 보기를 클릭하고 분석 섹션에서 Data Fusion > 인스턴스를 클릭합니다.

참고: 인스턴스를 만드는 데 20분 정도 걸립니다. 준비될 때까지 기다려 주세요.

이제 다음 단계를 사용하여 인스턴스와 연결된 서비스 계정에 권한을 부여합니다.

2. Google Cloud 콘솔에서 IAM 및 관리자 > IAM으로 이동합니다.

3. Compute Engine 기본 서비스 계정 {project-number}-compute@developer.gserviceaccount.com이 있는지 확인하고 서비스 계정을 클립보드에 복사합니다.

4. IAM 권한 페이지에서 +액세스 권한 부여를 클릭합니다.

5. 새 주 구성원 필드에 서비스 계정을 붙여넣습니다.

6. 역할 선택 필드를 클릭하고 Cloud Data Fusion API 서비스 에이전트를 입력한 다음 선택합니다.

7. 다른 역할 추가를 클릭합니다.

8. Dataproc 관리자 역할을 추가합니다.

9. 저장을 클릭합니다.

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다. 서비스 계정에 Cloud Data Fusion API 서비스 에이전트 역할 추가

서비스 계정 사용자 권한 부여

1. 콘솔의 탐색 메뉴에서 IAM 및 관리자 > IAM을 클릭합니다.

2. Google 제공 역할 부여 포함 체크박스를 선택합니다.

3. 목록을 아래로 스크롤하여 service-{project-number}@gcp-sa-datafusion.iam.gserviceaccount.com과 같은 Google 관리 Cloud Data Fusion 서비스 계정을 찾은 후 서비스 계정 이름을 클립보드에 복사합니다.

4. 다음으로 IAM 및 관리자 > 서비스 계정으로 이동합니다.

5. {project-number}-compute@developer.gserviceaccount.com과 같은 기본 Compute Engine 계정을 클릭하고 상단 탐색 메뉴에서 액세스 권한이 있는 주 구성원 탭을 선택합니다.

6. 액세스 권한 부여 버튼을 클릭합니다.

7. 새 주 구성원 필드에 앞서 복사한 서비스 계정을 붙여넣습니다.

8. 역할 드롭다운 메뉴에서 서비스 계정 사용자를 선택합니다.

9. 저장을 클릭합니다.

작업 7. Cloud Data Fusion UI 탐색

Cloud Data Fusion을 사용할 때는 Cloud Console과 별도의 Cloud Data Fusion UI를 모두 사용합니다. Cloud Console에서는 Cloud Console 프로젝트를 만들고, Cloud Data Fusion 인스턴스를 만들고 삭제할 수 있습니다. Cloud Data Fusion UI에서는 파이프라인 스튜디오 또는 Wrangler와 같은 다양한 페이지를 통해 Cloud Data Fusion 기능을 사용할 수 있습니다.

Cloud Data Fusion UI를 탐색하려면 다음 단계를 따르세요.

1. Cloud 콘솔에서 Data Fusion으로 돌아간 다음 Data Fusion 인스턴스 옆에 있는 인스턴스 보기 링크를 클릭합니다. 실습 사용자 인증 정보를 선택하여 로그인합니다. 서비스를 둘러볼지 묻는 메시지가 나타나면 아니요를 클릭합니다. 이제 Cloud Data Fusion UI가 나타날 것입니다.

2. Cloud Data Fusion 컨트롤 센터에서 탐색 메뉴를 사용하여 왼쪽 메뉴를 표시한 다음 파이프라인 > 스튜디오를 선택합니다.

3. 왼쪽 상단에서 드롭다운 메뉴를 사용하여 데이터 파이프라인 - 실시간을 선택합니다.

작업 8. 실시간 파이프라인 빌드

데이터로 작업할 때는 변환의 시작점으로 사용할 수 있도록 원시 데이터가 어떤 모습인지 확인할 수 있으면 편리합니다. 이를 위해 Wrangler를 사용하여 데이터를 준비하고 정리합니다. 이러한 데이터 중심 접근방식을 통해 변환을 빠르게 시각화할 수 있으며 실시간 피드백을 통해 올바른 방향으로 진행되고 있는지 확인할 수 있습니다.

1. 플러그인 팔레트의 변환 섹션에서 Wrangler를 선택합니다. Wrangler 노드가 캔버스에 표시됩니다. 속성 버튼을 클릭하여 엽니다.

2. 지시어 섹션 아래의 WRANGLE 버튼을 클릭합니다.

3. 로드가 완료되면 왼쪽 메뉴에서 업로드를 클릭합니다. 다음으로 업로드 아이콘을 클릭하여 이전에 컴퓨터에 다운로드한 샘플 트윗 파일을 업로드합니다.

4. 데이터가 Wrangler 화면에서 행/열 형식으로 로드됩니다. 몇 분 정도 걸릴 수 있습니다.

참고: 이를 Pub/Sub에서 최종적으로 수신하게 될 이벤트의 샘플로 취급합니다. 이는 파이프라인을 개발하는 동안 일반적으로 프로덕션 데이터에 액세스할 수 없는 실제 시나리오를 나타냅니다.

그러나 관리자가 작은 샘플에 대한 액세스 권한을 부여할 수 있습니다. 또는 API 계약을 준수하는 모의 데이터를 사용할 수 있습니다. 이 섹션에서는 각 단계에서 의견을 반영하여 이 샘플에 반복적으로 변환을 적용합니다. 그런 다음 실제 데이터에 변환을 재생하는 방법을 배우게 됩니다.

5. 첫 번째 작업은 JSON 데이터를 행과 열로 분할된 테이블 형식으로 파싱하는 것입니다. 이렇게 하려면 첫 번째 열(body) 제목에서 드롭다운 아이콘을 선택한 다음 파싱 메뉴 항목을 선택하고, 하위 메뉴에서 JSON을 선택합니다. 팝업에서 깊이를 1로 설정한 다음 적용을 클릭합니다.

   

6. 이전 단계를 반복하여 추가 변환을 위한 보다 의미 있는 데이터 구조를 확인합니다. body 열 드롭다운 아이콘을 클릭한 다음 파싱 > JSON을 선택하고 깊이를 1로 설정한 다음 적용을 클릭합니다.

   

   UI를 사용하는 것 외에도 Wrangler 지시어 명령줄 상자에 변환 단계를 작성할 수 있습니다. 이 상자는 Wrangler UI의 하단에 표시됩니다(녹색 $ 프롬프트가 있는 명령어 콘솔을 찾습니다). 다음 단계에서 명령어 콘솔을 사용하여 일련의 변환 단계를 붙여넣습니다.

7. 아래의 변환 단계를 모두 복사하여 Wrangler 지시어 명령줄 상자에 붙여넣어 추가합니다.

columns-replace s/^body_payload_//g drop id_str parse-as-simple-date :created_at EEE MMM dd HH:mm:ss Z yyyy drop display_text_range drop truncated drop in_reply_to_status_id_str drop in_reply_to_user_id_str parse-as-json :user 1 drop coordinates set-type :place string drop geo,place,contributors,is_quote_status,favorited,retweeted,filter_level,user_id_str,user_url,user_description,user_translator_type,user_protected,user_verified,user_followers_count,user_friends_count,user_statuses_count,user_favourites_count,user_listed_count,user_is_translator,user_contributors_enabled,user_lang,user_geo_enabled,user_time_zone,user_utc_offset,user_created_at,user_profile_background_color,user_profile_background_image_url,user_profile_background_image_url_https,user_profile_background_tile,user_profile_link_color,user_profile_sidebar_border_color,user_profile_sidebar_fill_color,user_profile_text_color,user_profile_use_background_image drop user_following,user_default_profile_image,user_follow_request_sent,user_notifications,extended_tweet,quoted_status_id,quoted_status_id_str,quoted_status,quoted_status_permalink drop user_profile_image_url,user_profile_image_url_https,user_profile_banner_url,user_default_profile,extended_entities fill-null-or-empty :possibly_sensitive 'false' set-type :possibly_sensitive boolean drop :entities drop :user_location 참고: '데이터가 없습니다. 변환 단계를 삭제해 보세요.'와 같은 메시지가 표시되면 X를 클릭하여 변환 단계 중 하나를 삭제하고 데이터가 표시되면 계속 진행할 수 있습니다.

8. 오른쪽 상단에 있는 적용 버튼을 클릭합니다. 그런 다음 오른쪽 상단에서 X를 클릭하여 속성 상자를 닫습니다.

보시다시피 파이프라인 스튜디오로 돌아왔으며, 캔버스에 단일 노드가 배치되어 있습니다. 이 노드는 Wrangler에서 정의한 변환을 나타냅니다. 그러나 위에서 설명한 것처럼 사용자 노트북의 데이터 대표 샘플에 이러한 변환을 적용했으며 실제 프로덕션 위치의 데이터에는 적용하지 않았기 때문에 이 파이프라인에 연결된 소스가 없습니다.

다음 단계에서는 데이터가 실제로 위치할 곳을 지정해 보겠습니다.

9. 플러그인 팔레트의 소스 섹션에서 PubSub를 선택합니다. PubSub 소스 노드가 캔버스에 표시됩니다. 속성 버튼을 클릭하여 엽니다.

10. 아래와 같이 PubSub 소스의 다양한 속성을 지정합니다.

a. 참조 이름에 Twitter_Input_Stream을 입력합니다.

b. 구독에서 cdf_lab_subscription(이전에 만든 PubSub 구독의 이름)을 입력합니다.

참고: PubSub 소스는 정규화된 구독 이름을 허용하지 않으며, .../subscriptions/ 부분 뒤의 마지막 구성요소만 허용합니다.

c. 검증을 클릭하여 오류가 없는지 확인합니다.

d. 오른쪽 상단에서 X를 클릭하여 속성 상자를 닫습니다.

11. 이제 PubSub 소스 노드를 이전에 추가한 Wrangler 노드에 연결합니다.

이전에 Wrangler에서 데이터 샘플을 사용했으므로 소스 열이 Wrangler에 body로 표시되었습니다. 그러나 PubSub 소스는 message라는 이름의 필드에서 이를 내보냅니다. 다음 단계에서 이 불일치를 수정합니다.

12. Wrangler 노드의 속성을 열고 기존 변환 단계의 상단에 다음 지시어를 추가합니다.

keep :message set-charset :message 'utf-8' rename :message :body

오른쪽 상단에서 X를 클릭하여 속성 상자를 닫습니다.

13. 이제 소스와 변환을 파이프라인에 연결했으므로 싱크를 추가하여 파이프라인을 완성합니다. 왼쪽 측면 패널의 싱크 섹션에서 BigQuery를 선택합니다. BigQuery 싱크 노드가 캔버스에 표시됩니다.

14. Wrangler 노드에서 BigQuery 노드로 화살표를 드래그하여 Wrangler 노드를 BigQuery 노드에 연결합니다. 다음으로 BigQuery 노드 속성을 구성합니다.

    

15. BigQuery 노드 위에 마우스를 가져간 후 속성을 클릭합니다.

    a. 참조 이름에 realtime_pipeline을 입력합니다.

    b. 데이터 세트에 realtime을 입력합니다.

    c. 테이블에 tweets를 입력합니다.

    d. 검증을 클릭하여 오류가 없는지 확인합니다.

16. 오른쪽 상단에서 X를 클릭하여 속성 상자를 닫습니다.

17. 파이프라인 이름 지정을 클릭하고 Realtime_Pipeline을 이름으로 추가한 다음 저장을 클릭합니다.

18. 배포 아이콘을 클릭한 다음 파이프라인을 시작합니다.

19. 배포가 완료되면 실행을 클릭합니다. 파이프라인 상태가 실행 중으로 변경될 때까지 기다립니다. 몇 분 정도 걸릴 수 있습니다.

작업 9. Cloud Pub/Sub에 메시지 전송

Dataflow 템플릿을 사용해 이벤트를 구독에 일괄 로드하여 이벤트를 전송합니다.

이제 템플릿을 기반으로 Dataflow 작업을 만들어 트윗 파일의 여러 메시지를 처리하고 앞서 만든 pubsub 주제에 게시합니다. Dataflow 작업 만들기 페이지에서 지속적으로 데이터 처리(스트리밍) 템플릿 아래의 Cloud Storage의 텍스트 파일을 Pub/Sub로를 사용합니다.

1. Cloud 콘솔로 돌아가서 탐색 메뉴로 이동하고 모든 제품 보기를 클릭한 후 분석 섹션에서 Dataflow를 클릭합니다.

2. 상단 메뉴 바에서 템플릿에서 작업 만들기를 클릭합니다.

3. Cloud Dataflow 작업의 작업 이름으로 streaming-pipeline을 입력합니다.

4. Cloud Dataflow 템플릿에서 Cloud Storage의 텍스트 파일을 Pub/Sub로 템플릿을 선택합니다.

5. Cloud Storage 입력 파일 아래에 gs://<YOUR-BUCKET-NAME>/<FILE-NAME>을 입력합니다. <YOUR-BUCKET-NAME>을 버킷 이름으로 바꾸고 <FILE-NAME>을 앞서 컴퓨터에 다운로드한 파일 이름으로 바꿉니다.

예: gs://qwiklabs-gcp-01-dfdf34926367/pubnub_tweets_2019-06-09-05-50_part-r-00000

6. 출력 Pub/Sub 주제에서 projects/<PROJECT-ID>/topics/cdf_lab_topic을 입력합니다.

PROJECT-ID를 실제 프로젝트 ID로 바꿔야 합니다.

7. 임시 위치에서 <YOUR-BUCKET-NAME>/tmp/를 입력합니다.

<YOUR-BUCKET-NAME>을 버킷 이름으로 바꿔야 합니다.

8. 작업 실행 버튼을 클릭합니다.

9. Dataflow 작업을 실행하고 몇 분 정도 기다립니다. pubsub 구독에서 메시지를 확인하고 실시간 CDF 파이프라인을 통해 처리되는 것을 볼 수 있습니다.

   

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다. 런타임 파이프라인 빌드 및 실행

작업 10. 파이프라인 측정항목 보기

이벤트가 Pub/Sub 주제에 로드되는 즉시 파이프라인에서 이벤트를 사용하기 시작합니다. 각 노드의 측정항목이 업데이트되는지 확인하세요.

• Data Fusion 콘솔에서 파이프라인 측정항목이 변경될 때까지 기다립니다.

  

수고하셨습니다.

이 실습에서는 Cloud Pub/Sub에서 스트리밍 수신 메시지를 읽고 데이터를 처리하여 BigQuery에 기록하는 Data Fusion의 실시간 파이프라인을 설정하는 방법을 배웠습니다.

설명서 최종 업데이트: 2025년 2월 6일

실습 최종 테스트: 2025년 2월 6일

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