개요

이 실습에서는 다음 작업을 수행합니다.

• 브랜치가 있는 파이프라인 구현
• 쓰기 전에 데이터 필터링
• 데이터를 <Row> 객체로 처리
• 파이프라인에 커스텀 명령줄 파라미터 추가

기본 요건

Java에 대한 기본 지식

이전 실습에서는 기본 추출-변환-로드 순차 파이프라인을 만들고 이에 상응하는 Dataflow 템플릿을 사용하여 Google Cloud Storage에 일괄 데이터 스토리지를 수집했습니다. 이 파이프라인은 일련의 변환으로 구성됩니다.

하지만 많은 파이프라인이 이처럼 단순한 구조를 나타내지는 않습니다. 이 실습에서는 더 정교한 비순차적 파이프라인을 빌드합니다.

여기서의 사용 사례는 리소스 소비를 최적화하는 것입니다. 제품마다 리소스를 소비하는 방식이 다릅니다. 또한 비즈니스 내에서 모든 데이터가 동일한 방식으로 사용되는 것은 아닙니다. 예를 들어 분석 워크로드 내에서 일부 데이터는 정기적으로 쿼리되는 반면, 일부 데이터는 복구에만 사용됩니다.

이 실습에서는 첫 번째 실습의 파이프라인을 리소스 소비에 맞게 최적화합니다. 분석가가 사용할 데이터만 BigQuery에 저장하고 다른 데이터는 매우 저렴하고 내구성이 뛰어난 스토리지 서비스인 Google Cloud Storage의 Coldline Storage에 보관합니다.

설정 및 요건

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

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

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. 저장을 클릭합니다.

IDE 설정

이 실습에서는 Google Compute Engine에서 호스팅되는 Theia Web IDE를 주로 사용합니다. 실습 저장소가 사전에 클론되어 있습니다. Java 언어 서버가 지원되며, Cloud Shell처럼 gcloud 명령줄 도구를 통해 Google Cloud API에 프로그래매틱 방식으로 액세스할 수 있는 터미널도 제공됩니다.

1. Theia IDE에 액세스하려면 Google Skills에 표시된 링크를 복사하여 새 탭에 붙여넣습니다.

참고: 환경이 완전히 프로비저닝되려면 URL이 표시된 후에도 3~5분 정도 기다려야 할 수 있습니다. 환경이 완전히 프로비저닝되지 않으면 브라우저에 오류가 표시됩니다.

실습 저장소가 환경에 클론되었습니다. 각 실습은 사용자가 완성해야 하는 코드가 포함된 labs 폴더와, 문제 발생 시 참고할 수 있는 정상 작동 예시가 포함된 solution 폴더로 구분되어 있습니다.

2. File Explorer 버튼을 클릭하여 다음을 확인합니다.

Cloud Shell을 사용할 때처럼 이 환경에서 여러 터미널을 만들 수도 있습니다.

터미널에서 gcloud auth list를 실행하면, 제공된 서비스 계정으로 로그인되어 있음을 확인할 수 있습니다. 이 서비스 계정은 실습 사용자 계정과 동일한 권한을 가지고 있습니다.

환경이 작동하지 않는다면, GCE 콘솔에서 IDE를 호스팅하는 VM을 다음과 같이 재설정할 수 있습니다.

여러 변환으로 동일한 PCollection 처리

이 실습에서는 데이터를 Google Cloud Storage와 BigQuery에 모두 쓰는 브랜치 파이프라인을 작성합니다. 브랜치 파이프라인을 작성하는 한 가지 방법은 동일한 PCollection에 서로 다른 두 개의 변환을 적용하여 서로 다른 두 개의 PCollection을 생성하는 것입니다.

[PCollection1] = [Initial Input PCollection].apply([A Transform]) [PCollection2] = [Initial Input PCollection].apply([A Different Transform])

브랜치 파이프라인 구현

이 섹션이나 이후 섹션에서 문제가 발생하면 해결 방법을 참고하세요.

작업 1. Cloud Storage에 쓰기 위한 브랜치 추가

이 작업을 완료하려면 Cloud Storage에 쓰는 브랜치를 추가하여 기존 파이프라인을 수정합니다.

적절한 실습 열기

• 아직 IDE 환경에 새 터미널을 만들지 않았다면 새 터미널을 만들고 다음 명령어를 복사하여 붙여넣습니다.

# Change directory into the lab cd 2_Branching_Pipelines/labs # Download dependencies mvn clean dependency:resolve export BASE_DIR=$(pwd)

데이터 환경 설정

# Create GCS buckets and BQ dataset cd $BASE_DIR/../.. source create_batch_sinks.sh # Generate event dataflow source generate_batch_events.sh # Change to the directory containing the practice version of the code cd $BASE_DIR

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다. 데이터 환경 설정

1. IDE에서 2_Branching_Pipelines/labs/src/main/java/com/mypackage/pipeline에 있는 MyPipeline.java를 엽니다.
2. 아래로 스크롤하여 파이프라인의 본문이 정의된 run() 메서드를 찾습니다. 이 메서드는 현재 다음과 같습니다.

pipeline.apply("ReadFromGCS", TextIO.read().from(input)) .apply("ParseJson", ParDo.of(new JsonToCommonLog())) .apply("WriteToBQ", BigQueryIO.<CommonLog>write().to(output).useBeamSchema() .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE) .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

3. 각 요소가 Json에서 <CommonLog>로 변환되기 전에 TextIO.write()를 사용하여 Cloud Storage에 쓰는 새로운 브랜치 변환을 추가하여 이 코드를 수정합니다.

이 섹션이나 이후 섹션에서 문제가 발생하면 training-data-analyst 파일에서 해결 방법을 참고하세요.

스키마를 사용하는 이유 먼저 살펴 보기

스키마는 특정 프로그래밍 언어 유형과 관계없이 Beam 레코드의 유형 시스템을 제공합니다. 모두 동일한 스키마를 가진 여러 Java 클래스(예: Protocol-Buffer 클래스 또는 POJO 클래스)가 있을 수 있으며 Beam을 사용하면 이러한 유형 간에 원활하게 변환할 수 있습니다. 또한 스키마는 다양한 프로그래밍 언어 API에서 유형을 추론하는 간단한 방법을 제공합니다.

스키마가 있는 PCollection은 Coder를 지정할 필요가 없습니다. Beam이 스키마 행을 인코딩하고 디코딩하는 방법을 알고 있기 때문입니다. Beam은 스키마 유형을 인코딩하기 위해 특수한 코더를 사용합니다. Schema API가 도입되기 전에는 Beam이 파이프라인의 모든 객체를 인코딩하는 방법을 알고 있어야 했습니다.

작업 2. 필드별 데이터 필터링

현재는 모든 데이터가 두 번 저장되므로 새 파이프라인이 실제로 리소스를 덜 소비하지는 않습니다. 리소스 소비를 개선하려면 중복된 데이터의 양을 줄여야 합니다. Google Cloud Storage 버킷은 보관 및 백업 스토리지로 작동하도록 설계되었으므로 모든 데이터를 여기에 저장하는 것이 중요합니다. 하지만 모든 데이터를 BigQuery로 보낼 필요는 없습니다.

데이터 분석가가 웹사이트에서 사용자가 액세스하는 리소스와, 사용자의 액세스 패턴이 지역과 시간에 따라 어떻게 다른지 자주 살펴본다고 가정해 보겠습니다. 필드의 하위 집합만 필요합니다.

각 객체를 변환하고 필드의 하위 집합만 반환하는 DoFn을 작성할 수도 있지만, Apache Beam은 스키마가 있는 PCollection에 대해 다양한 관계형 변환을 제공합니다. 각 레코드가 이름이 지정된 필드로 구성되어 있기 때문에 SQL 표현식의 집계와 유사하게 이름으로 필드를 참조하는 간단하고 읽기 쉬운 집계가 가능합니다.

Select 및 DropFields는 이러한 두 가지 변환입니다.

PCollection<MyClass> pCollection = ...; pCollection.apply("SelectUserState", Select.fieldNames("state"))); PCollection<MyClass> pCollection = ...; pCollection.apply("DropPII", DropFields.fields("ssn", "state"));

중요: 각 예시는 PCollection<MyClass> 대신 PCollection<Row>을 반환합니다. Row 클래스는 모든 스키마를 지원할 수 있으며 스키마화된 일반 객체로 간주될 수 있습니다. 스키마가 있는 모든 PCollection은 행의 PCollection으로 변환될 수 있습니다.

위의 두 변환 모두 필드가 삭제되므로 전체 CommonLog 객체를 반환하지 않으며, 따라서 변환은 Row를 반환하도록 되돌아갑니다. 새로운 이름이 지정된 스키마를 만들거나 중간 POJO 스키마를 등록할 수도 있지만, 당분간은 Row를 사용하는 것이 더 쉽습니다.

• 이 작업을 완료하려면 다음 import 문을 추가하고, 파이프라인에 이러한 변환 중 하나를 추가하여 분석가가 사용하려는 필드만 전송하도록 BigQuery에 저장되는 필드 집합을 변경합니다.

import org.apache.beam.sdk.values.Row; import org.apache.beam.sdk.schemas.transforms.Select; import org.apache.beam.sdk.schemas.transforms.DropFields;

기억할 사항: BigQueryIO.<CommonLog>write() 메서드가 이미 연결되어 있다면 새 유형으로 인해 이 메서드를 <Row>로 변경해야 합니다.

작업 3. 요소별 데이터 필터링

Apache Beam에는 다양한 필터링 방법이 있습니다. 이전 작업에서는 스키마 변환을 사용하는 한 가지 방법을 보여드렸습니다. 이 구현에서는 각 요소의 일부를 필터링하여 스키마와 필드의 하위 집합이 남아 있는 새로운 Row 객체를 생성했습니다. 아래 예시처럼 전체 요소를 필터링하는 데에도 똑같이 쉽게 사용할 수 있습니다.

purchases.apply(Filter.<MyObject>create() .whereFieldName(“costCents”, (Long c) -> c > 100 * 20) .whereFieldName(“shippingAddress.country”, (String c) -> c.equals(“de”)); 참고: 이 필터 변환(org.apache.beam.sdk.schemas.transforms.Filter)을 이전의 비스키마 필터 함수(org.apache.beam.sdk.transforms.Filter)와 혼동해서는 안 됩니다.

• 이 작업을 완료하려면 먼저 아래의 import 문을 코드에 추가한 다음 파이프라인에 필터 변환을 추가합니다. 원하는 기준에 따라 필터링할 수 있습니다. 람다 함수에 유형 힌트(예: (Integer c) -> c > 100)를 추가해야 할 수도 있습니다.

import org.apache.beam.sdk.schemas.transforms.Filter;

작업 4. 커스텀 명령줄 파라미터 추가

현재 파이프라인에는 입력 경로와 BigQuery의 테이블 위치를 포함하여 여러 파라미터가 하드 코딩되어 있습니다. 하지만 파이프라인이 Cloud Storage의 모든 JSON 파일을 읽을 수 있다면 더 유용할 것입니다. 이 특성을 추가하려면 명령줄 파라미터 세트에 추가해야 합니다.

현재 파이프라인은 PipelineOptionsFactory를 사용하여 Options라는 커스텀 클래스 인스턴스를 생성하지만 Options 클래스는 PipelineOptions 클래스와 아무런 차이가 없으므로 사실상 PipelineOptions의 인스턴스입니다.

public interface Options extends PipelineOptions { } public static void main(String[] args) { Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class); run(options); }

PipelineOptions 클래스는 다음 형식을 따르는 명령줄 인수를 해석합니다.

--<option>=<value>

하지만 이는 사전 정의된 파라미터의 작은 집합입니다. 여기에서 get- 함수를 확인할 수 있습니다.

커스텀 파라미터를 추가하려면 두 가지 작업을 수행해야 합니다.

1. 먼저 아래 예시와 같이 Options 클래스에 상태 변수를 추가합니다.

public interface Options extends PipelineOptions { @Description("My custom command line argument.") @Default.String("DEFAULT") String getMyCustomOption(); void setMyCustomOption(String myCustomOption); }

2. 둘째, main() 메서드 내에서 PipelineOptionsFactory로 인터페이스를 등록한 다음 PipelineOptions 객체를 만들 때 인터페이스를 전달합니다. PipelineOptionsFactory로 인터페이스를 등록하면 --help는 커스텀 옵션 인터페이스를 찾고 --help 명령어 출력에 추가할 수 있습니다. 또한 PipelineOptionsFactory는 커스텀 옵션이 등록된 모든 옵션과 호환되는지 확인합니다.

다음 예시 코드는 PipelineOptionsFactory로 커스텀 옵션 인터페이스를 등록하는 방법을 보여줍니다.

PipelineOptionsFactory.register(Options.class); Options options = PipelineOptionsFactory.fromArgs(args) .withValidation() .as(Options.class);

3. 코드에서 명령줄 파라미터에 액세스하려면 파라미터의 get 함수를 호출하기만 하면 됩니다.

String myCustomOption = option.getMyCustomOption();

4. 이 작업을 완료하려면 먼저 다음 import 문을 추가한 다음 입력 경로, Google Cloud Storage 출력 경로, BigQuery 테이블 이름에 대한 명령줄 파라미터를 추가하고 파이프라인 코드를 업데이트하여 상수 대신 이러한 파라미터에 액세스합니다.

import org.apache.beam.sdk.options.Default; import org.apache.beam.sdk.options.Description;

작업 5. 파이프라인에 NULLABLE 필드 추가

지난 실습에서 만든 BigQuery 테이블에는 다음과 같이 모든 필드가 REQUIRED인 스키마가 있었습니다.

파이프라인 실행 자체와 이를 반영하는 스키마가 있는 결과 BigQuery 테이블 모두에서 데이터가 누락된 경우 NULLABLE 필드가 있는 Apache Beam 스키마를 만드는 것이 좋습니다.

Javax 표기법을 클래스 정의에 추가할 수 있으며, 그러면 다음과 같이 Apache Beam 스키마에 통합됩니다.

@DefaultSchema(JavaFieldSchema.class) class MyClass { int field1; @javax.annotation.Nullable String field2; }

• 이 작업을 완료하려면 클래스 정의에서 lat 및 lon 필드를 NULLABLE로 표시합니다.

작업 6. 명령줄에서 파이프라인 실행

• 이 작업을 완료하려면 명령줄에서 파이프라인을 실행하고 적절한 파라미터를 전달합니다. NULLABLE 필드의 BigQuery 스키마 결과를 기록해 두세요. 코드는 다음과 같이 표시됩니다.

# Set up environment variables export PROJECT_ID=$(gcloud config get-value project) export REGION={{{ project_0.default_region | "REGION" }}} export BUCKET=gs://${PROJECT_ID} export COLDLINE_BUCKET=${BUCKET}-coldline export PIPELINE_FOLDER=${BUCKET} export MAIN_CLASS_NAME=com.mypackage.pipeline.MyPipeline export RUNNER=DataflowRunner export INPUT_PATH=${PIPELINE_FOLDER}/events.json export OUTPUT_PATH=${PIPELINE_FOLDER}-coldline export TABLE_NAME=${PROJECT_ID}:logs.logs_filtered cd $BASE_DIR mvn compile exec:java \ -Dexec.mainClass=${MAIN_CLASS_NAME} \ -Dexec.cleanupDaemonThreads=false \ -Dexec.args=" \ --project=${PROJECT_ID} \ --region=${REGION} \ --stagingLocation=${PIPELINE_FOLDER}/staging \ --tempLocation=${PIPELINE_FOLDER}/temp \ --runner=${RUNNER} \ --inputPath=${INPUT_PATH} \ --outputPath=${OUTPUT_PATH} \ --tableName=${TABLE_NAME}" 참고: 파이프라인이 성공적으로 빌드되지만 Dataflow 서비스의 코드 또는 잘못된 구성으로 인해 많은 오류가 발생하는 경우 RUNNER를 'DirectRunner'로 다시 설정하여 로컬에서 실행하고 더 빠른 피드백을 받을 수 있습니다. 이 경우에는 데이터 세트가 작고 DirectRunner에서 지원하지 않는 특성을 사용하지 않기 때문에 이 방식이 적합합니다.

작업 7. 파이프라인 결과 확인

1. Dataflow 작업 페이지로 이동하여 작업이 실행되는 모습을 살펴봅니다. 그래프는 다음과 유사합니다.

2. Filter 함수를 나타내는 노드를 클릭합니다. 위의 그림에서는 FilterFn으로 표시되어 있습니다. 오른쪽에 표시되는 패널에서 출력으로 작성된 것보다 더 많은 요소가 입력으로 추가되었음을 확인할 수 있습니다.

3. 이제 Cloud Storage에 쓰기를 나타내는 노드를 클릭합니다. 모든 요소가 작성되었으므로 이 숫자는 필터 함수의 입력에 있는 요소의 수와 일치해야 합니다.

4. 파이프라인이 완료되면 테이블을 쿼리하여 BigQuery에서 결과를 검토합니다. 테이블의 레코드 수는 Filter 함수에서 출력된 요소의 수와 일치해야 합니다.

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다. 명령줄에서 파이프라인 실행

실습 종료하기

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

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

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

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

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

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

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