GCP 서비스 이해 및 실습

2019-04-17

Data Engineering

Data_Engineering_TIL_(20190416)

study program : https://www.fastcampus.co.kr/extension_des

[학습목표]

GCP(Google Cloud Platform) 데이터 분석용 서비스 이해
GCP(Google Cloud Platform) 서비스 실습

[학습기록]

번외로 현업관련 팁

작년과 비교해 마이크로소프트 Azure가 확연히 좋아지고 있다.
Airflow을 적용 시 데이터파이프라인 구축하는데 장점이 많음.
최근 온프레미스에서는 도커와 쿠버네티스 조합이 두드러집니다(온프레미스 상에서의 리소스 관리 목적으로 쿠버네티스를 많이 사용), 환경구성에 매우 편리하고, 프로젝트 빌드에 매우 효율적이다.
ML시 데이터 전처리가 가장 중요하고 힘든데 최근에는 도커를 활용하는 것이 정말 좋다.
클라우드 운영비용은 아키텍처를 구성하는 엔지니어가 어떻게 하느냐에 따라서 전적으로 좌지우지 된다. 이런 아키텍처를 어떻게 잘 구성할 것인가를 항상 고민해야한다.
통상 타블로서버가 컴퓨팅 파워를 많이 먹고, 사용자가 눈으로 보는 영역이기 때문에 EC2 리소스를 사양이 높은 것을 쓸 수 밖에 없다.
통상 스토리지 비용은 1년에 그렇게 많이 드는 편은 아니다. 보통 휴대폰 서비스 기업들은 데이터를 분석해서 쌓아둔 것만 1테라바이트 정도 든다. 로우데이터들은 따로 빼내서 glacier 얼려놓는다.
기본적으로 빅데이터 분석을 하면 고민해야 할 사항

1) 컴퓨팅에 대한 고민

AWS에서 EC2가 컴퓨팅하는데 기본단위이다.

구글에도 이와 유사한 개념인 compute engine이라는 것이 있다.

Temporary virtual machines으로 AWS에는 Spot instances가 있는데 구글에 유사한 개념으로 Preemptible VMs가 있다.

Firewall도 AWS의 Security groups과 유사한 개념으로 구글에는 Compute Engine firewall rules이 있다.

2) 데이터가 들어오는 네트워크에 대한 고민

(백본망에 대한 이해)

3) storage에 대한 고민

(어디에 저장을 할 것인지)

4) insight를 어떻게 찾아낼 것인가에 대한 고민

AWS 키네시스와 Google pub/sub

AWS 키네시스와 유사한 서비스로 구글에는 google pub/sub이라는게 있다. 구글펍섭은 단점이 있는데 데이터를 처리하는 순서가 없다. 스트림하고 큐를 이용해서 그런지 데이터가 들어오는 순서대로 처리하는 것을 보장하지 않는다. 그리고 구글펍섭도 키네시스와 마찬가지로 데이터가 들어오면 한군대로 들어오는데 컨슈머를 여러개 두고 흘려줄수는 있음. 구글 펍섭만의 장점은 큐에서 이미지 같은 큰 데이터도 처리가 가능함. 키네시스보다 10배정도 큰 데이터들을 처리가능함. 그러나 데이터를 처리하는데 순서가 중요한만큼 순서가 섞이는 것은 안좋은 단점이다. 반면에 키네시스는 들어오는 순서에 맞게 차례차례 처리함.

실습프리뷰

실습목적 :

DataProc를 이용해서 house recommendation 을 구현하는 실습을 수행한다.

집에 대한 정보가 나와있는 데이터와 집을 구매하고 rating한 정보가 담겨 있는 정보를 collaborative filtering 알고리즘으로 머신러닝해서 recommendation spark 엔진으로 DataProc로 돌려서 결과값을 sql디비에 넣을 것이다.

학습된 결과는 예를 들어서 특정사용자 id가 5번인 사용자는 다음에 어떤집을 추천해주면 좋겠다 라는 것을 찾는것이다.

이를 위해서 아래 그림과 같은 자원들을 사용해볼 것이고, 아키텍처를 구성해볼 것이다.

step 1) GCP 접속 및 cloud shell 사용해보기

1) 콘솔 확인

개별 서비스 완성도가 높은 반면 전체 구조의 짜임새는 좀 더 AWS에 비해 자유로운 느낌)

2) Shell Open

클라우드 쉘은 AWS에는 없는 기능인데

클라우드 쉘은 일종에 aws 인스턴스와 같다고 보면 된다.

웹에서 CLI환경으로 인스턴스 컴퓨팅을 할 수 있다.

step 2) Compute Engine(AWS에서 EC2와 유사) 실습

1) Create Compute Engine Instance

2) ssh 접속 후 아래 코드 실행

sudo yum update –y

sudo yum install httpd -y

sudo yum update -y

sudo systemctl start httpd.service

ps -ef

grep httpd 명령어로 프로세스 정상실행여부 확인

sudo touch /var/www/html/index.html 명령어 실행

sudo chmod 777 index.html 명령어 실행

sudo echo "<html><h1> google cloud test </h1></html>" > /var/www/html/index.html 명령어 실행

인스턴스 외부아이피 확인하여 웹브라우저로 해당아이피로 접속하여 잘 접속이 되는지 확인

3) Web Service External IP 로 접속 후 확인

4) Delete the instance

step 3) Storage(AWS S3와 유사) 실습

1) 버킷 생성

2) 클라우드 쉘에서 git을 통해 repo clone

git clone https://github.com/GoogleCloudPlatform/training-data-analyst

3) clone한 폴더를 gsutil을 통해 버킷으로 copy

step 4) SQL 실습

1) Instance Create

2) Import sql & csv

sql : ~/training-data-analyst/CPB100/lab3a/cloudsql/table_creation.sql

csv :

~/training-data-analyst/CPB100/lab3a/cloudsql/accommodation.csv

~/training-data-analyst/CPB100/lab3a/cloudsql/rating.csv

(database와 table 명을 위의 sql 파일에 따라 설정해줌)

3) find_my_ip.sh 실행하여 콘솔 ip를 얻음

~/training-data-analyst/CPB100/lab3a/find_my_ip.sh

4) Connection 네트워크에 추가

5) 현재 내가 있는 오피스의 ip 얻기

6) Connection 네트워크에 추가

7) MySQL 개요에서 데이터 베이스 ip를 얻어 로컬 SQLworkbench로 접속테스트

8) Web에서 Connect Using Cloud shell을 통해 접속

step 5) DataProc(AWS EMR과 유사) for house recommendation 실습

dataproc는 참고로 리소스와 job을 분리해놓은 것이 특징이다.

1) DataProc Cluster 생성

주의사항으로 위의 SQL과 Region 및 Zone이 동일해야한다.

2) ~/training-data-analyst/CPB100/lab3b/authorize_dataproc.sh 의 CLOUDSQL을 자신의 인스턴스명으로 수정

3) ~/training-data-analyst/CPB100/lab3b/authorize_dataproc.sh <리젼 및="" 존명=""> <워커노드 수=""> 실행

4) ~/training-data-analyst/CPB100/lab3b/sparkml/train_and_apply.py 의 CLOUDSQL_INSTANCE_IP, CLOUDSQL_PWD 수정

5) 위의 train_and_apply.py gsutil로 버킷에 복사

6) Job Submit

7) MySQL Client로 MySQL 인스턴스에 접속하여 Recommendation 테이블 확인

job이 잘 돌았는지 확인

#### 실습 상세내용

instance를 구동해서 웹페이지를 띄우는 실습

step1) 클라우드 쉘 활성화를 먼저해보자. 구글 클라우드 플렛폼 콘솔로 접속해서 우측상단에 클라우드 쉘 활성화를 클릭한다.

그러면 아래와 같이 CLI환경의 콘솔이 전시될 것이다.

나의 구글 클라우드 플랫폼의 도스 콘솔에서 명령어로 무언가 리소스를 컨트롤 할 수 있도록, 이렇게 웹콘솔로 지원해주고 있다.

AWS에서 유사한 기능으로는 AWS CLI라는 것이 있다.

step2) 그 다음에 본격적으로 instance를 띄워보려고 한다. 좌측메뉴창에서 컴퓨트 엔진을 클릭해준다.

그 다음에 VM 인스턴스 메뉴에서 만들기 클릭

step 3) 만들기 옵션에서 아래 그림과 같이 옵션선택

부팅디스크는 센트오에스로 해준다. 참고로 리눅스는 우분투를 제일 많이 쓰는데 무거운 것이 단점이다. 센트오에스가 리눅스 버전중에는 가벼운 편이다. 결론적으로 개인이 사용하기에는 우분투가 좋은데 서버용은 센트오에스로 하는것이 좋다.

step 4) 인스턴스를 생성하면 다음과 같은 화면이 전시가 되고 우리는 여기서 SSH를 클릭해준다.

SSH를 클릭해주면 아래 그림처럼 웹상의 콘솔로 띄워주게 된다.

AWS EC2 접속한 것과 같이 리눅스 터미널이 열리면서 전시가 될 것이다.

step 5) 아래 코드들을 실행해준다.

sudo yum update –y

(최신 라이브러리로 업데이트)

yum은 패키지 저장소를 수요되는 패키지를 검색하고 다운로드 하여 설치까지 처리하면서 패키지들의 의존성을 고려하여 설치할때 패키지 의존성에 대한 error를 줄일 수 있다. Red Hat Linux에서는 yum으로 패키지를 설치한다.

sudo yum install httpd -y

(웹서버 설치)

sudo yum update -y

sudo systemctl start httpd.service

(웹서버를 데몬프로그램에 등록)

ps -ef

grep httpd

(http 서비스 프로세스 정상실행여부 확인)

sudo touch /var/www/html/index.html

(index.html 파일 생성)

sudo chmod 777 index.html

(write 권한이 없어서 권한을 주는 작업)

sudo echo "<html><h1> google cloud test </h1></html>" > /var/www/html/index.html 명령어 실행

(index.html 파일안에 google cloud test 텍스트 삽입)

step 6) 아래 그림처럼 인스턴스의 지정된 외부아이피로 접속하여 우리가 입력한 html 문자열이 잘 전시가 되는지 확인한다.

13-2

step 7) 이렇게 확인이 되면 만든 인스턴스를 삭제해준다.

그 다음에 Storage(AWS S3와 유사)를 실습해보자.

아래 그림과 같이 다시 콘솔에서 메뉴 클릭 -> 스토리지 클릭

step 8) 아래 그림과 같은 옵션으로 버켓생성

step 9) 버킷이 생성되면 버킷이 생성된 것을 아래 그림처럼 확인할 수 있다.

우리는 house recommendation을 위해 실습할 데이터를 가져올 것이다.

데이터는 깃허브 서버에서 가져올 것이고, 클라우드 쉘을 거쳐서 생성한 버킷에 넣어줄 것이다.

(데이터를 가져올 깃허브 주소 : https://github.com/GoogleCloudPlatform/training-data-analyst )

이 클라우드 쉘도 인스턴스와 결국에는 유사한것이기 때문에 일정량의 디스크가 있다.

또한 이 콘솔에 깃허브 패키지가 깔려있어서 git clone 같은 명령어를 인식할 수 있다.

그래서 이 클라우드 쉘을 거쳐서 버킷으로 실습데이터를 보낼 것이다.

그래서 우리는 클라우드 쉘에서 아래 그림과 같은 깃클론 명령어를 입력해 데이터를 다운로드 받아준다.

step10) 그다음에 아래 그림의 빨간색 박스처럼 명령어를 입력해서 실행해준다.

구글 클라우드는 CLI 명령어로 버킷에 데이터를 업로드할때 gsutil이라는 명령어를 제공한다.

아래 그림의 gsutil 명령어는 minman2115라는 구글클라우드에 training-data-analyst 폴더 전체를 복사해서 보내주겠다라는 의미이다.

-r 옵션(서브폴더까지 다 포함하겠다라는 의미)

완료가 되면 아래 그림처럼 버킷 서브폴더에서 데이터를 확인할 수 있다.

step11) 그 다음에 우리는 구글클라우드 mysql 서비스를 띄워보려고한다.

마찬가지로 아래 그림처럼 메뉴에서 sql을 클릭해준다.

그 다음에 인스턴스 생성을 클릭해준다.

그 다음에 그러면 인스턴스 만들기가 나오는데 mysql 선택해준다.

그런 다음에 MySQL 2세대 인스턴스 만들기가 나오는데 아래 그림과 같은 옵션으로 생성해준다.

그러면 아래 그림처럼 SQL 인스턴스가 생성된 것을 확인할 수 있다.

step12) 이제 버킷에 있는 데이터를 방금생성한 SQL 디비에 올리려고 한다.

디비에 데이터들이 정상적으로 올라가는지는 mysql workbench 툴로 확인할 것이다.

버킷에서 디비로 보내줄 파일은 총 세개이다.

~/training-data-analyst/CPB100/lab3a/cloudsql 경로에 있는

table_creation.sql, accommodation.csv, rating.csv 총 세개이다.

(accommodation.csv : 주거지에 대한 정보를 갖고 있는 데이터, rating.csv : 주거지에 대한 점수 데이터)

sql 인스턴스에서 내 인스턴스 아이디클릭 -> 화면상단에 가져오기 클릭 -> 클라우드 스토리지 파일에서 찾아보기 클릭 -> 위의 경로로 들어가서 -> 먼저 table_creation.sql 파일 선택 -> 가져오기 형식은 sql -> 가져오기 클릭

(참고로 table_creation.sql가져올때 가져오기 형식에서 데이터베이스 옵션은 그냥 놔두면된다. 기존에 데이터베이스를 생성해 놓은게 아니기 때문이다. 그리고 소스안에 creation하는 내용이 있다. recommendation spark이라는 데이터베이스를 생성하고 그 밑에다가 테이블을 생성할 것이다.)

마찬가지로 accommodation.csv, rating.csv를 같은 방법으로 가져오는데 가져오기 형식만 아래 그림처럼 맞춰주면 된다.

그렇게해주면 SQL에서 아래 sql파일을 참고하여 파일을 가져와서 데이터베이스를 생성하고, 테이블까지 자동으로 구성해준다.

참고로 table_creation.sql의 내용은 아래와 같다.

CREATE DATABASE IF NOT EXISTS recommendation_spark;

USE recommendation_spark;

DROP TABLE IF EXISTS Recommendation;
DROP TABLE IF EXISTS Rating;
DROP TABLE IF EXISTS Accommodation;
CREATE TABLE IF NOT EXISTS Accommodation
(
  id varchar(255),
  title varchar(255),
  location varchar(255),
  price int,
  rooms int,
  rating float,
  type varchar(255),
  PRIMARY KEY (ID)
);

CREATE TABLE  IF NOT EXISTS Rating
(
  userId varchar(255),
  accoId varchar(255),
  rating int,
  PRIMARY KEY(accoId, userId),
  FOREIGN KEY (accoId) 
    REFERENCES Accommodation(id)
);

CREATE TABLE  IF NOT EXISTS Recommendation
(
  userId varchar(255),
  accoId varchar(255),
  prediction float,
  PRIMARY KEY(userId, accoId),
  FOREIGN KEY (accoId) 
    REFERENCES Accommodation(id)
);

step13) sql 접근을 위해 클라우드 쉘과 지금 있는 오피스의 아이피를 알아내서 sql에 등록해줘야한다.

그래서 먼저 클라우드 쉘에서 아래의 경로로 가서 ./find_my_ip.sh을 실행해서 퍼블릭 아이피를 알아낸다.

(클라우드 쉘 아이피 획득)

참고로 클라우드 쉘의 아이피를 sql에 등록해주는 이유는 이따가 추천하는 연산을 수행한 결과를 다시 디비에다가 넣어줄 것인데 그때 필요하다.

step14) 그 다음에 웹브라우져를 하나 열어서 www.myip.com 에 접속하여 현재 오피스의 아이피 획득

step15) sql 인스턴스 세부정보에서 ‘연결’메뉴로 들어간다음에 아래 그림과 같이 방금알아낸 클라우드 쉘 아이피와 지금 있는 오피스의 아이피를 네트워크 추가를 클릭해서 등록해준다.

참고로 만약에 클라우드 내의 서비스끼리 등록을 한다고 하면, private 아이피로 세팅해줘야 한다. 만약에 클라우드 내의 서비스끼리 등록하는건데 퍼블릭 아이피로 등록을 해주면 외부로 나갔다가 다시 들어오게 된다.

step16) sql워크밴치로 생성한 sql 인스턴스에 이제 접속해보려고 한다.

먼저 sql 인스턴스 세부정보로 들어가면 ‘이 인스턴스에 연결’이라는 메뉴에 sql 인스턴스 접근을 위한 아이피주소가 나와 있을것이다. 그걸 카피를 해주고, sql 워크밴치를 실행한 다음에 host name에 카피한 아이피 주소를 복사해주고, sql 비빌먼호등을 아래 그림과 같이 입력해준 다음에 접속을 한다.

step17) 디비로 접속해서 우리가 넣은 csv 파일들이 잘 들어가 있나 확인해보고, 아래와 같은 sql 쿼리로 데이터를 조회해본다.

use recommendation_spark;

show tables;

select *

from Accommodation;

step18) 그 다음스텝으로 data proc로 클러스터를 구성해보려고 한다.

이 data proc을 통해서 sql에 있는 테이블 데이터를 가져다가 data proc이 일종에 collaborative filtering을 알고리즘을 돌려서 그것의 결과를 다시 sql에 넣어주는 작업을 할 것이다.

마찬가지로 아래 그림과 같이 메뉴에서 data proc를 클릭해준다.

step19) 아래 그림과 같은 옵션으로 클러스터를 생성해준다.

step20) 이렇게 클러스터를 구성하고 그 다음에 잡을 등록할 것이다. 그러기 위해 사전에 여러 설정작업을 해줘야 한다.

먼저 아래의 그림처럼 authorize_dataproc.sh가 있는 디렉토리로 가서 authorize_dataproc.sh를 vim으로 실행한 다음에 CLOUDSQL 부분을 내 클라우드 sql 아이디로 변경을 해주고, authorize_dataproc 실행으로 안에 서술된 내용을 적용시킨다.

(./authorize_dataproc.sh 클러스터아이디 지역 작업자수)

이렇게하면 클러스터(마스터노드와 슬레이브노드)가 디비에 접근할 수 있도록 아이피 설정을 수행해준다.

접근가능하도록 이렇게 설정을 해주면 이 노드들이 sql에서 데이터를 가져와서 job을 한다음에 결과를 또 sql에 써줄 것이다.

그 다음에 train_and_apply.py가 있는 디렉토리로가서 train_and_apply.py를 vim으로 실행한 다음에 일부 수정을 해야한다. data proc가 sql에서 데이터를 읽고, 결과를 sql에다가 써야하기 때문에 클라우드sql 아이피 부분과, sql 비밀번호를 입력해서 sql에 접근이 가능한 형태로 만들어줘야 한다. train_and_apply.py은 그밖에 실질적으로 어떤 작업을 해야하는지 명시되어 있다.

그런 다음에 gsutil cp ./train_and_apply.py gs://minman2115 명령어를 실행해준다.

(이렇게 설정이 적용된 train_and_apply.py를 버킷에다가 카피해서 올려준다. data proc에서 읽을 수 있는 gs스토리지 안에 있는 것만 읽을 수 있기 때문에 스토리지로 올려주는 것이다.)

참고로 아래에 코드는 train_and_apply.py 내용이다.

참고로 train_and_apply.py 코드는 아래와 같다.

#!/usr/bin/env python
"""
Copyright Google Inc. 2016
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""


import os
import sys
import pickle
import itertools
from math import sqrt
from operator import add
from os.path import join, isfile, dirname
from pyspark import SparkContext, SparkConf, SQLContext
from pyspark.mllib.recommendation import ALS, MatrixFactorizationModel, Rating
from pyspark.sql.types import StructType, StructField, StringType, FloatType

CLOUDSQL_INSTANCE_IP = '104.155.188.32'   # CHANGE (database server IP)
CLOUDSQL_DB_NAME = 'recommendation_spark'
CLOUDSQL_USER = 'root'
CLOUDSQL_PWD  = 'root'  # CHANGE

conf = SparkConf().setAppName("train_model")
sc = SparkContext(conf=conf)
sqlContext = SQLContext(sc)

jdbcDriver = 'com.mysql.jdbc.Driver'
jdbcUrl    = 'jdbc:mysql://%s:3306/%s?user=%s&password=%s' % (CLOUDSQL_INSTANCE_IP, CLOUDSQL_DB_NAME, CLOUDSQL_USER, CLOUDSQL_PWD)

# checkpointing helps prevent stack overflow errors
sc.setCheckpointDir('checkpoint/')

# Read the ratings and accommodations data from Cloud SQL
dfRates = sqlContext.read.format('jdbc').options(driver=jdbcDriver, url=jdbcUrl, dbtable='Rating', useSSL='false').load()
dfAccos = sqlContext.read.format('jdbc').options(driver=jdbcDriver, url=jdbcUrl, dbtable='Accommodation', useSSL='false').load()
print("read ...")

# train the model
model = ALS.train(dfRates.rdd, 20, 20) # you could tune these numbers, but these are reasonable choices
print("trained ...")

# use this model to predict what the user would rate accommodations that she has not rated
allPredictions = None
for USER_ID in range(0, 100):
  dfUserRatings = dfRates.filter(dfRates.userId == USER_ID).rdd.map(lambda r: r.accoId).collect()
  rddPotential  = dfAccos.rdd.filter(lambda x: x[0] not in dfUserRatings)
  pairsPotential = rddPotential.map(lambda x: (USER_ID, x[0]))
  predictions = model.predictAll(pairsPotential).map(lambda p: (str(p[0]), str(p[1]), float(p[2])))
  predictions = predictions.takeOrdered(5, key=lambda x: -x[2]) # top 5
  print("predicted for user={0}".format(USER_ID))
  if (allPredictions == None):
    allPredictions = predictions
  else:
    allPredictions.extend(predictions)

# write them
schema = StructType([StructField("userId", StringType(), True), StructField("accoId", StringType(), True), StructField("prediction", FloatType(), True)])
dfToSave = sqlContext.createDataFrame(allPredictions, schema)
dfToSave.write.jdbc(url=jdbcUrl, table='Recommendation', mode='overwrite')

step21) 클러스터에 작업을 명령하는 것은 클러스터 메뉴에서 ‘작업’ 메뉴를 클릭해주면 된다. 작업 메뉴로 들어가서 ‘작업 제출’을 클릭하고 아래와 같은 옵션을 설정한 다음에 작업을 제출하면 클러스터가 train_and_apply.py에 명시되어 있는 작업들을 실시할 것이다.

아래 그림에서 볼 수 있듯이 ‘기본 python 파일’을 설정하는 옵션이 있는데 아까 우리가 gsutil cp ./train_and_apply.py gs://minman2115해서 붙어넣은 train_and_apply.py를 말한다. 일종의 작업지시서 같은 것이다.

step22) 그리고 mysql workbench로 다시 sql에 접속을해서 작업 결과에 대해 확인을 해보면 된다.

아래와 같은 sql 쿼리를 입력해서 결과를 확인해보자

결과는 아래 그림과 같다.

5번 사용자에 대해 추천해주는 집의 데이터를 보여줌

select     r.userid, 
              r.accoid, 
              r.prediction, 
              a.title, 
    a.location, 
    a.price, 
    a.rooms, 
    a.rating, 
    a.type  
from Recommendation as r 
join Accommodation as a 
on r.accoid = a.id 
 and r.userid = 5;

 스텍프레임 기초개념 CS TIL (20180813) 