개요

이 실습에서 학습할 내용은 다음과 같습니다.

• 사용자별로 사이트 트래픽을 합산하는 파이프라인 작성
• 분 단위로 사이트 트래픽을 합산하는 파이프라인 작성
• 시계열 데이터에 윈도잉 구현

기본 요건

Java에 대한 기본 지식

설정 및 요건

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

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

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

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

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

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

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

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

프로젝트 권한 확인

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

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 개요 가이드를 참조하세요.

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을 다음과 같이 재설정할 수 있습니다.

1부: 사용자별 사이트 트래픽 합산

실습의 이 부분에서는 다음 작업을 수행하는 파이프라인을 작성합니다.

1. Cloud Storage의 파일에서 하루치 트래픽을 읽습니다.
2. 각 이벤트를 CommonLog 객체로 변환합니다.
3. 각 객체를 사용자 ID별로 그룹화하고 값을 결합하여 순 사용자 각각의 조회수를 합산하고 해당 사용자의 총 조회수를 구합니다.
4. 각 사용자에 대해 추가 합산을 수행합니다.
5. 결과 데이터를 BigQuery에 작성합니다.

작업 1. 합성 데이터 생성

이전 실습과 마찬가지로 첫 번째 단계는 파이프라인에서 처리할 데이터를 생성하는 것입니다. 이전과 동일하게 실습 환경을 열고 데이터를 생성합니다.

적절한 실습 열기

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

# Change directory into the lab cd 3_Batch_Analytics/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

이 스크립트는 다음과 유사한 줄이 포함된 events.json이라는 파일을 만듭니다.

{"user_id": "-6434255326544341291", "ip": "192.175.49.116", "timestamp": "2019-06-19T16:06:45.118306Z", "http_request": "\"GET eucharya.html HTTP/1.0\"", "lat": 37.751, "lng": -97.822, "http_response": 200, "user_agent": "Mozilla/5.0 (compatible; MSIE 7.0; Windows NT 5.01; Trident/5.1)", "num_bytes": 182}

그러면 이 파일이 gs://my-project-id/events.json에 있는 Google Cloud Storage 버킷에 자동으로 복사됩니다.

• Google Cloud Storage로 이동하여 스토리지 버킷에 events.json이라는 파일이 있는지 확인합니다.

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

작업 2. 사용자당 페이지 조회수 합계

1. IDE에서 3_Batch_Analytics/labs/src/main/java/com/mypackage/pipeline에 있는 BatchUserTrafficPipeline.java를 엽니다.

이 파이프라인에는 입력 경로와 출력 테이블 이름에 대한 명령줄 옵션을 수락하는 데 필요한 코드와 Google Cloud Storage에서 이벤트를 읽고, 이벤트를 파싱하고, 결과를 BigQuery에 쓰는 코드도 이미 포함되어 있습니다. 하지만 몇 가지 중요한 부분이 누락되었습니다.

2. 파이프라인의 다음 단계는 각 고유 user_id별로 이벤트를 합산하고 각 이벤트의 페이지 조회수를 계산하는 것입니다. Beam 스키마가 있는 Rows 또는 객체에서 이 작업을 수행하는 쉬운 방법은 Group.byFieldNames() 변환을 사용한 다음, 결과 그룹에 대해 몇 가지 합산을 수행하는 것입니다. 예를 들면 다음과 같습니다.

purchases.apply(Group.<MyObject>byFieldNames("userId", "address"));

'key'와 'values'라는 두 개의 필드가 있는 행의 PCollection을 반환합니다. 'key' 필드 자체는 userID와 address의 모든 고유한 조합을 나타내는 <userID:STRING, address:STRING> 스키마를 가진 Row입니다. 'values' 필드는 해당 고유 그룹의 모든 객체를 포함하는 ITERABLE[ROW[MyObject]] 유형입니다.

필드 이름	필드 유형
key	ROW{userId:STRING, streetAddress:STRING}
values	ITERABLE[ROW[Purchase]]

3. 이 기능은 다음과 같이 이 그룹화에 대한 합산을 수행하고 결과 필드에 이름을 지정할 수 있을 때 가장 유용합니다.

purchases.apply(Group.byFieldNames("userId") .aggregateField("itemId", Count.combineFn(), "numPurchases") .aggregateField("costCents", Sum.ofLongs(), "totalSpendCents") .aggregateField("costCents", Max.ofLongs(), "largestPurchases"));

복수형인 'values'가 아닌 'key' 'value' Row를 반환합니다.

필드 이름	필드 유형
key	ROW{userId:STRING}
value	ROW{numPurchases: INT64, totalSpendCents: INT64, largestPurchases: INT64}

Sum 및 Count 변환은 이러한 용도에 적합합니다. Sum 및 Count는 데이터 그룹에 작용할 수 있는 Combine 변환의 예입니다.

참고: 이 예시에서는 이 변환이 전체 그룹에 있는 요소의 개수를 세는 것이므로 Count.combineFn()의 필드나 와일드카드 필드*에서 합산할 수 있습니다.

4. 파이프라인의 다음 단계는 user_id별로 이벤트를 합산하고, 페이지 조회수를 합산하며, num_bytes에 대한 몇 가지 추가 합산(예: 총 사용자 바이트)을 수행하는 것입니다.

이 작업을 완료하려면 파이프라인에 다른 변환을 추가하여 user_id별로 이벤트를 그룹화한 다음, 관련 합산을 수행합니다. 입력, 사용할 CombineFn, 출력 필드 이름을 지정하는 방법을 염두에 두세요.

작업 3. 스키마 평탄화

이 시점에서 새 변환은 이미 언급한 것처럼 스키마 <Key,Value>를 사용하여 PCollection을 반환합니다. 파이프라인을 그대로 실행하면 각 행에 기본적으로 하나의 값만 있어도 BigQuery에 두 개의 중첩된 RECORD로 작성됩니다.

1. 다음과 같은 Select 변환을 추가하여 이를 방지할 수 있습니다.

purchases.apply(Group.byFieldNames("userId") .aggregateField("itemId", Count.combineFn(), "numPurchases") .aggregateField("costCents", Sum.ofLongs(), "totalSpendCents") .aggregateField("costCents", Max.ofLongs(), "largestPurchases")) .apply(Select.fieldNames("key.userId", "value.numPurchases", "value.totalSpendCents", "value.largestPurchases")

이렇게 하면 새로운 평탄화된 스키마에 관련 필드 이름이 유지되고 'key'와 'value'가 삭제됩니다.

2. 이 작업을 완료하려면 Select 변환을 추가하여 새 행의 스키마를 평탄화합니다.

참고: 아직 변경하지 않았다면 BigQueryIO.<CommonLog>write()의 객체 힌트를 <Row>로 변경해야 합니다.

작업 4. 파이프라인 실행

• Cloud Shell로 돌아가서 다음 명령어를 실행하여 Dataflow 서비스를 사용해 파이프라인을 실행합니다. 문제가 있는 경우 DirectRunner로 실행하거나 솔루션을 참고할 수 있습니다.

export PROJECT_ID=$(gcloud config get-value project) export REGION={{{ project_0.default_region | "REGION" }}} export BUCKET=gs://${PROJECT_ID} export PIPELINE_FOLDER=${BUCKET} export MAIN_CLASS_NAME=com.mypackage.pipeline.BatchUserTrafficPipeline export RUNNER=DataflowRunner export INPUT_PATH=${PIPELINE_FOLDER}/events.json export TABLE_NAME=${PROJECT_ID}:logs.user_traffic 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} \ --tableName=${TABLE_NAME}"

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다. 사용자별로 사이트 트래픽을 합산하고 파이프라인 실행

작업 5. BigQuery에서 결과 확인

1. 이 작업을 완료하려면 파이프라인이 완료될 때까지 몇 분 정도 기다린 다음 BigQuery로 이동하여 user_traffic 테이블을 쿼리합니다.

2. 궁금하다면 Select 변환 단계를 주석 처리하고 파이프라인을 다시 실행하여 결과 BigQuery 스키마를 확인해 보세요.

2부: 분 단위로 사이트 트래픽 합산

실습의 이 부분에서는 BatchMinuteTraffic이라는 새 파이프라인을 만듭니다. BatchMinuteTraffic은 BatchUserTraffic에서 사용한 기본 배치 분석 원칙을 확장하여 전체 배치에서 사용자를 기준으로 합산하는 대신 이벤트가 발생한 시점을 기준으로 합산합니다.

• IDE에서 3_Batch_Analytics/labs/src/main/java/com/mypackage/pipeline 내의 BatchMinuteTrafficPipeline.java 파일을 엽니다.

작업 1. 각 요소에 타임스탬프 추가

제한되지 않은 소스는 각 요소에 타임스탬프를 제공합니다. 제한되지 않은 소스에 따라 원시 데이터 스트림에서 타임스탬프를 추출하는 방법을 구성해야 할 수 있습니다.

그러나 이 파이프라인에서 사용되는 TextIO의 파일과 같은 제한된 소스는 타임스탬프를 제공하지 않습니다.

1. 각 레코드에서 타임스탬프 필드를 파싱하고 WithTimestamps 변환을 사용해 PCollection의 각 요소에 타임스탬프를 연결할 수 있습니다.

PCollection<MyClass> unstamped = ...; PCollection<MyClass> stamped = unstampedLogs.apply(WithTimestamps.of((MyClass m) -> org.joda.time.Instant.parse(m.getTimestamp())));

2. 이 작업을 완료하려면 파이프라인의 각 요소에 타임스탬프를 추가하는 변환을 파이프라인에 추가합니다.

작업 2. 1분 기간으로 윈도잉

윈도잉은 개별 요소의 타임스탬프에 따라 PCollection을 세분화합니다. GroupByKey, Combine과 같은 여러 요소를 합산하는 변환은 암시적으로 기간별로 작동합니다. 즉, 전체 컬렉션의 크기가 제한되지 않은 경우에도 각 PCollection을 여러 개의 연속적인 유한 기간으로 처리합니다.

여러 유형의 기간을 정의하여 PCollection의 요소를 나눌 수 있습니다. Beam은 다음을 비롯한 여러 윈도우 함수를 제공합니다.

• 고정 시간 기간
• 슬라이딩 시간 기간
• 세션별 기간
• 단일 전역 기간
• 캘린더 기반 기간(Python용 Beam SDK에서 지원되지 않음)

이 실습에서는 고정 시간 기간을 사용합니다. 고정 시간 기간은 데이터 스트림에서 시간 간격이 겹치지 않는 일정한 지속 시간을 나타냅니다. 5분 길이의 기간을 고려해 보겠습니다. 타임스탬프 값이 0:00:00부터 0:05:00(0:05:00 제외)까지인 제한되지 않은 PCollection의 모든 요소는 첫 번째 기간에 속하고, 타임스탬프 값이 0:05:00부터 0:10:00(0:10:00 제외)까지인 요소는 두 번째 기간에 속하는 식입니다.

1. 다음과 같이 1초 길이의 고정 시간 기간을 구현합니다.

PCollection<String> pColl= ...; PCollection<String> windowedPCollection = pColl.apply( Window.<String>into(FixedWindows.of(Duration.standardSeconds(60))));

2. 이 작업을 완료하려면 요소를 1분 길이의 기간으로 윈도잉하는 변환을 파이프라인에 추가합니다.

다른 유형의 윈도잉에 대해 자세히 알아보려면 Apache Beam 문서의 윈도우 함수 섹션을 참고하세요.

작업 3. 기간당 이벤트 수 계산

다음으로 파이프라인은 각 기간 내에서 발생한 이벤트 수를 계산해야 합니다. BatchUserTraffic 파이프라인에서는 키별로 합산하기 위해 Sum 변환이 사용되었습니다. 그러나 이 경우에는 요소가 윈도잉되었으며 원하는 계산 방식이 기간 경계를 준수해야 한다는 점에서 이전 파이프라인과 다릅니다.

이러한 새로운 제약 조건에도 불구하고 Combine 변환은 여전히 적절합니다. Combine 변환은 자동으로 기간 경계를 준수하기 때문입니다.

1. 기간당 요소 수를 계산하는 새로운 변환을 추가하는 방법은 Count 문서를 참고하세요.

Beam 2.22 기준으로 윈도잉 중에 행의 요소를 계산하는 가장 좋은 방법은 완전한 SQL을 사용하지 않고 Combine.globally(Count.<T>combineFn()).withoutDefaults()를 사용하는 것입니다. 완전한 SQL 사용은 다음 실습에서 자세히 다룰 예정입니다. 이 변환은 PCollection<Long> 유형을 출력하며, 더 이상 Beam 스키마를 사용하지 않습니다.

2. 이 작업을 완료하려면 각 기간의 모든 요소를 계산하는 변환을 추가합니다. 문제가 발생하면 솔루션을 참고하세요.

작업 4. 다시 행으로 변환하고 타임스탬프 추가

BigQuery에 쓰려면 각 요소를 'pageviews' 필드와 'minute'라는 추가 필드를 포함하는 Row 객체로 다시 변환해야 합니다. 이는 각 기간의 경계를 하나의 필드로 사용하고 페이지 조회수의 합계를 다른 필드로 사용하는 방식입니다.

지금까지는 JSON 문자열에서 CommonLog 객체로 변환된 요소는 항상 Beam 스키마를 준수했으며, 때로는 Row 객체로 되돌아가는 경우도 있었습니다. 원래 스키마는 @DefaultSchema(JavaFieldSchema.class) 주석을 통해 CommonLog POJO에서 추론되었으며, 이후에 추가/삭제된 필드는 파이프라인 변환에 지정되었습니다. 그러나 파이프라인의 이 시점에서 Count 변환의 출력에 따라 모든 요소는 Long 유형입니다. 따라서 새로운 Row 객체를 처음부터 만들어야 합니다.

1. 다음과 같이 스키마를 수동으로 만들고 등록할 수 있습니다. 이 코드는 CommonLog 객체 정의와 마찬가지로 main() 메서드 외부에 추가됩니다.

// Define the schema for the records. Schema appSchema = Schema .builder() .addInt32Field("appId") .addStringField("description") .addDateTimeField("rowtime") .build();

2. 이 스키마의 후속 Row 객체는 Long과 같은 입력을 기반으로 PTransform에서 생성될 수 있습니다.

Row row = Row .withSchema(appSchema) .addValues(1, "Some cool app", new Date()) .build();

3. 변환이 이전 행을 변경하는 것이 아니라 새 행을 만드는 경우 Beam은 일반적으로 PTransform에 새 스키마를 표시해야 합니다.

.apply().setRowSchema(appSchema)

Apache의 Beam SQL 안내에서 스키마 생성과 추론에 대해 자세히 알아보세요.

이 시점에서 또 다른 문제는 Count 변환이 더 이상 타임스탬프 정보를 포함하지 않는 Long 유형의 요소만 제공한다는 것입니다.

하지만 실제로 요소는 타임스탬프 정보를 포함합니다. 명확하게 드러나지 않을 뿐입니다. Apache Beam 실행기는 기본적으로 이벤트 타임스탬프, 기간, 파이프라인 옵션 등 여러 추가 파라미터의 값을 제공하는 방식을 지원합니다. 전체 목록은 Apache의 DoFn 파라미터 문서를 참고하세요.

4. 이 작업을 완료하려면 Long 유형의 요소를 수락하고 제공된 pageViewsSchema 스키마 유형을 사용하여 Row 유형의 요소를 내보내며 IntervalWindow 유형의 추가 입력 파라미터를 포함하는 ParDo 함수를 작성합니다. 이 추가 파라미터를 사용하여 Instant 인스턴스를 만들고 이를 사용하여 분 필드의 문자열 표현을 도출합니다.":"

@ProcessElement public void processElement(@Element T l, OutputReceiver<T> r, IntervalWindow window) { Instant i = Instant.ofEpochMilli(window.start().getMillis()); ... r.output(...); }

작업 5. 파이프라인 실행

• 코딩을 완료한 후 아래 명령어를 사용하여 파이프라인을 실행합니다. 코드를 테스트하는 동안 파이프라인을 로컬에서 실행하는 DirectRunner로 RUNNER 환경 변수를 변경하면 훨씬 더 빠르게 테스트할 수 있습니다.

export PROJECT_ID=$(gcloud config get-value project) export REGION={{{ project_0.default_region | "REGION" }}} export BUCKET=gs://${PROJECT_ID} export PIPELINE_FOLDER=${BUCKET} export MAIN_CLASS_NAME=com.mypackage.pipeline.BatchMinuteTrafficPipeline export RUNNER=DataflowRunner export INPUT_PATH=${PIPELINE_FOLDER}/events.json export TABLE_NAME=${PROJECT_ID}:logs.minute_traffic 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} \ --tableName=${TABLE_NAME}"

내 진행 상황 확인하기를 클릭하여 목표를 확인합니다. 분 단위로 사이트 트래픽을 합산하고 파이프라인 실행

작업 6. 결과 확인

• 이 작업을 완료하려면 파이프라인이 실행될 때까지 몇 분 정도 기다린 다음 BigQuery로 이동하여 minute_traffic 테이블을 쿼리합니다.

실습 종료하기

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

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

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

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

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

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

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