지속가능한 데이터 분석을 위한 AWS 환경구축 및 활용 실습

2019-05-10

Data Engineering

Data_Engineering_TIL_(20190509)

study program : https://www.meetup.com/awskrug

[학습시 참고자료]

실습관련 URL

1) 핸즈온 자료

https://github.com/awskrug/datascience-group/tree/master/workshop-sustainable_data_analysis

2) AWS 기반 지속 가능한 데이터 분석 플랫폼 구축하기 (AWS Summit Seoul 2019)

https://www.slideshare.net/awskorea/aws-aws-summit-seoul-2019-141290115

[학습프리뷰]

1) EMR을 이용한 분석환경 구축

2) Zeppelin을 이용한 데이터 전처리

3) AWS Glue를 이용한 데이터 카탈로그 생성

4) Athena에서 카탈로그를 통해 데이터 로딩

5) Tableau를 이용하여 데이터 시각화

[실습 아키텍처]

아래와 같은 서비스 아키텍처를 가정하여 데이터분석을 위한 엔지니어링 실습

ETL 이란

ETL은 Extract, Transform, Load의 각 앞글자를 따온 것이며, 아래의 과정을 포괄하는 개념이다.

Extract : 정제되지 않은(raw) 형태의 로그 데이터를 추출하여,

Transform : 모니터링이나 분석에 적합한 형태의 데이터로 변환하고,

Load : 변환된 데이터를 새로운 DB나 스토리지에 적재한다.

위와 같은 데이터 ETL 과정을 통해 적재된 데이터는 현업에서 아래와 같은 용도로 활용할 수 있다.

비즈니스 의사결정에 필요한 주요 데이터를 빠르게 조회 및 탐색하는 분석 환경을 운영, 지표 모니터링에 필요한 대시보드를 생성 등

[실습 상세내용]

1. EMR을 이용한 분석환경 구축

step 1) 아마존 웹서비스 사용을 위한 키페어 생성

EMR 클러스터에 ssh 접속을 위해 EC2키페어를 생성한다.

콘솔 좌측의 키페어 메뉴로 접속하여 아래 그림과 같이 임의의 키페어 이름을 입력하고 키페어를 생성해준다.

1-2

step2) EMR 시작하기

콘솔에서 EMR 서비스로 이동한다음에 ‘클러스터 생성’을 클릭한다.

EMR 클러스터 생성을 위한 환경설정 메뉴를 확인하실 수 있는데 아래와 같이 옵션을 선택해주면 된다.

빅데이터 분석을 위해서 어떤 소프트웨어를 사용할 것인지 선택할 수 있다.

데이터 메터 데이터에서 AWS Glue 데이터 카탈로그 사용 이란 옵션이 있는데 추후에 Glue를 통해 생성한 데이터 카탈로그의 데이터들에 접근할 수 있다.

하드웨어를 선택할 수 있다. 선택하는 인스턴스에 따라 퍼포먼스와 비용에 영향을 줍니다. 또한 좌측 상단에 보면 고급 옵션으로 이동이란 메뉴가 있어서 원하는 환경을 직접 선택하여 구성할 수 있다.

고급 옵션으로 이동을 선택한다.

step3) EMR 구성하기

소프트웨어 구성을 위한 화면이다.

이전에 보았던 메뉴구성과 달리 원하시는 소프트웨어의 조합을 마음대로 구성할 수 있다.

Zeppelin과 Spark 환경을 구성하기 위해 Hadoop 2.8.5, Zeppelin 0.8.1, Spark 2.4.0을 선택한다.

AWS Glue 데이터 카탈로그 설정도 체크한다.

하드웨어 구성은 아래 그림처럼 마스터노드 1개는 온디맨드 형태로, 코어노드 2개는 스팟 형태로 띄운다.

주의해야 할점은 마스터노드는 반드시 온디맨드 형태로 해야한다는 것이다. 왜냐하면 마스터노드를 스팟형태로 하면 스팟자원이 다른사람에게 넘어가버리면 클러스터 전체가 다운되어버리기 때문이다.

마스터노드도 스팟인스턴스로 띄워도 되긴 하지만 마스터노드에 문제가 생기면 많은 귀찮고 어려운점들이 있으니 가능하면 온디맨드로 설정해준다.

마스터노드에는 Zeppelin을 포함한 여러가지 환경설정들이 저장되어 있어서, 마스터노드가 꺼질경우 전체 분석환경이 비활성화 된다고 생각하면 된다.

반면에 실제 데이터들을 분산하여 처리하는 코어노드나 작업노드는 원하는 작업양 만큼 자유롭게 늘이고 줄이고하는 작업이 가능합니다. 따라서 코어노드나 작업노드는 비용을 고려하여 스팟 인스턴스 형태로 띄워도 무방하다.

3-2

클러스터 이름을 임의로 입력하고 다음으로 넘어간다.

3-3

마지막 보안을 위한 설정인데 생성한 본인의 키페어를 선택하고 클러스터 생성 버튼을 누른다.클러스터 생성완료까지 수분이 소요된다.

3-4

EMR 클러스터 생성이 완료되면 아래와 같은 화면을 볼 수 있다.

마스터 퍼블릭 DNS는 Zeppelin이나 ssh 접속 시 사용할 수 있다.

ElasticMapReduce-master 보안그룹에서 보안그룹을 설정하여 접근권한을 제어할 수 있는데, 실습을 위해 현재 IP 기준으로 접근을 허가할 것이다.

그래서 아래 그림의 우측하단의 ‘마스터보안 그룹’ URL을 클릭한다.

아래 그림과 같이 마스터노드를 클릭하고 EC2 콘솔 보안 그룹 메뉴로 이동하여 보안그룹의 인바운드을 편집을 누른다.

4-2

아래 그림과 같이 제플린으로의 접근을 위해서 8890 포트를 열고 및 내 IP로 설정해준다. 그리고 ssh 접속을 위해서 22 포트 를 열고 내 IP로 설정해준다.

소스에서 내IP를 선택하면 자동으로 나의 아이피가 입력된다.

아래 그림과 같이 설정해주고 저장을 눌러준다.

그러면 정상적으로 Spark 마스터노드 안에 구성된Zeppelin에 접근이 가능한 상태까지 온것이다.

step4) Zeppelin으로 데이터를 분석하기 위한 간단한 설정작업

먼저 아래 그림의 좌측상단에 빨간색 박스의 SSH를 클릭해준다.

아래 그림처럼 DNS 주소를 확인한다.

Git Bash 터미널에서 키페어 있는 폴더로 이동한 다음에 아래 그림과 같이 마스터노드로 접속해준다.

접속한 다음에 아래그림과 같이 wget https://raw.githubusercontent.com/awskrug/datascience-group/master/workshop-sustainable_data_analysis/env_emr_spark_zeppelin.sh 명령어를 입력하여 제플린 설정관련 스크립트 파일을 다운받는다.

참고로 다운받은 스크립트 파일내용은 아래와 같다.

#!/bin/bash
    
echo "============================== ADD NEW USER FOR ZEPPELIN =============================="

# 새로운 사용자 추가
sudo useradd -m ds_handson_20190509

# 새로운 사용자의 암호 설정
sudo passwd ds_handson_20190509

# 새로운 사용자의 HDFS 디렉토리 생성
hadoop fs -mkdir /user/ds_handson_20190509

# 새로운 사용자의 HDFS 디렉토리 권한 수정
hadoop fs -chmod 777 /user/ds_handson_20190509

echo "============================== INSTALL AND UPDATE PACKAGES =============================="

# 필요한 패키지 업데이트
sudo yum update -y
sudo yum install python36 -y
sudo yum install python36-setuptools -y
sudo yum install python36-pip -y

# 필요한 python3 패키지 설치
sudo python3 -m pip install --upgrade pip
sudo python3 -m pip install numpy==1.16.3
sudo python3 -m pip install scipy==1.2.1
sudo python3 -m pip install pytz==2019.1
sudo python3 -m pip install six==1.12.0
sudo python3 -m pip install python-dateutil==2.8.0
sudo python3 -m pip install pandas==0.24.2
sudo python3 -m pip install scikit-learn==0.20.3
sudo python3 -m pip install matplotlib==3.0.3
sudo python3 -m pip install boto3
sudo python3 -m pip install google-api-python-client
sudo python3 -m pip install oauth2client

echo "============================== SET PYSPARK CONF =============================="

# python3 버전 확인
python3 -c "import sys;print(sys.version)"

# echo 명령 수행을 위해 spark-env.sh 파일에 쓰기 권한 임시 부여
sudo chmod 666 /etc/spark/conf.dist/spark-env.sh

# spark-env.sh 파일에 공백 줄 추가
echo "" >> /etc/spark/conf.dist/spark-env.sh
echo "" >> /etc/spark/conf.dist/spark-env.sh

# spark-env.sh 파일에 PYSPARK_PYTHON=python3 옵션 추가
echo "export PYSPARK_PYTHON=python3" >> /etc/spark/conf.dist/spark-env.sh

# spark-env.sh 파일 권한 복원
sudo chmod 644 /etc/spark/conf.dist/spark-env.sh

# spark-env.sh 파일 실행
source /etc/spark/conf.dist/spark-env.sh

echo "============================== SET ZEPPELIN CONF =============================="

# 설정 변경 전, Zeppelin 종료
sudo stop zeppelin

# /etc/zeppelin/conf.dist/zeppelin-site.xml 파일 생성 (zeppelin-site.xml.template 파일 복사)
sudo cp /etc/zeppelin/conf.dist/zeppelin-site.xml.template /etc/zeppelin/conf.dist/zeppelin-site.xml

# zeppelin을 로그인 없이 사용할 수 없도록 설정 (/etc/zeppelin/conf.dist/zeppelin-site.xml 수정)
sudo perl -i~ -0pe 's/zeppelin.anonymous.allowed<\/name>\n\s\s<value>true/zeppelin.anonymous.allowed<\/name>\n  <value>false/g' /etc/zeppelin/conf.dist/zeppelin-site.xml

# zeppelin 출력 사이즈 제한 조정 (102400bytes -> 1024000bytes)
sudo perl -i~ -0pe 's/zeppelin.interpreter.output.limit<\/name>\n\s\s<value>102400/zeppelin.interpreter.output.limit<\/name>\n  <value>1024000/g' /etc/zeppelin/conf.dist/zeppelin-site.xml

# echo 명령 수행을 위해 zeppelin-site.xml 파일에 쓰기 권한 임시 부여 (644 -> 666)
sudo chmod 666 /etc/zeppelin/conf.dist/zeppelin-site.xml

# property 추가를 위한 </configuration> 태그 임시 제거
sudo perl -i~ -0pe 's/<\/configuration>//g' /etc/zeppelin/conf.dist/zeppelin-site.xml

# zeppelin-site.xml 파일에 공백 줄 추가
echo "" >> /etc/zeppelin/conf.dist/zeppelin-site.xml
echo "" >> /etc/zeppelin/conf.dist/zeppelin-site.xml

# cron 작업 설정이 가능하도록 zeppelin.notebook.cron.enable: true 속성 추가
echo "<property>" >> /etc/zeppelin/conf.dist/zeppelin-site.xml
echo "  <name>zeppelin.notebook.cron.enable</name>" >> /etc/zeppelin/conf.dist/zeppelin-site.xml
echo "  <value>true</value>" >> /etc/zeppelin/conf.dist/zeppelin-site.xml
echo "</property>" >> /etc/zeppelin/conf.dist/zeppelin-site.xml

# </configuration> 태그 추가
echo "" >> /etc/zeppelin/conf.dist/zeppelin-site.xml
echo "</configuration>" >> /etc/zeppelin/conf.dist/zeppelin-site.xml

# zeppelin-site.xml 권한 복원 (666 -> 644)
sudo chmod 644 /etc/zeppelin/conf.dist/zeppelin-site.xml

# /etc/zeppelin/conf.dist/shiro.ini 파일 생성 (shiro.ini.template 파일 복사)
sudo cp /etc/zeppelin/conf.dist/shiro.ini.template /etc/zeppelin/conf.dist/shiro.ini

# 로그인한 모든 zeppelin 이용자에게 interpreter 설정 및 재시작 권한 부여 (/etc/zeppelin/conf.dist/shiro.ini 수정)
sudo perl -pi -e 's/authc, roles\[admin\]/authc/g' /etc/zeppelin/conf.dist/shiro.ini

# 사용하지 않는 인증 요소 비활성화
sudo perl -pi -e 's/admin = password1, admin/#admin = password1, admin/g' /etc/zeppelin/conf.dist/shiro.ini
sudo perl -pi -e 's/user1 = password2, role1, role2/#user1 = password2, role1, role2/g' /etc/zeppelin/conf.dist/shiro.ini
sudo perl -pi -e 's/user2 = password3, role3/#user2 = password3, role3/g' /etc/zeppelin/conf.dist/shiro.ini
sudo perl -pi -e 's/user3 = password4, role2/#user3 = password4, role2/g' /etc/zeppelin/conf.dist/shiro.ini

# pam 기반 사용자 인증 활성화
sudo perl -pi -e 's/#pamRealm=org.apache.zeppelin.realm.PamRealm/pamRealm=org.apache.zeppelin.realm.PamRealm/g' /etc/zeppelin/conf.dist/shiro.ini
sudo perl -pi -e 's/#pamRealm.service=sshd/pamRealm.service=sshd/g' /etc/zeppelin/conf.dist/shiro.ini

# zeppelin에서 마스터 인스턴스의 사용자 암호를 확인할 수 있도록 /etc/shadow 파일에 읽기 권한 부여
sudo setfacl -m user:zeppelin:r /etc/shadow

echo "============================== SET ZEPPELIN INTERPRETER CONF =============================="

# /etc/zeppelin/conf.dist/interpreter.json 수정 (value: python인 값을 value: python3로 수정)
sudo perl -pi -e 's/"value": "python"/"value": "python3"/g' /etc/zeppelin/conf.dist/interpreter.json

# zeppelin.spark.uiWebUrl 제거하고, spark.yarn.executor.memoryOverhead 추가
sudo perl -i~ -0pe 's/"zeppelin.spark.uiWebUrl":\s{\n\s\s\s\s\s\s\s\s\s\s"name":\s"zeppelin.spark.uiWebUrl",\n\s\s\s\s\s\s\s\s\s\s"value":\s"",\n\s\s\s\s\s\s\s\s\s\s"type":\s"string"\n\s\s\s\s\s\s\s\s}/"spark.yarn.executor.memoryOverhead": {\n          "name": "spark.yarn.executor.memoryOverhead",\n          "value": "4096",\n          "type": "number"\n        }/g' /etc/zeppelin/conf.dist/interpreter.json

sudo start zeppelin

echo "============================== END OF JOB =============================="

그 다음에는 아래 그림과 같이 sh env_emr_spark_zepplin.sh 명령어를 입력하여 다운받은 스크립트 파일을 실행한다.

그러면 Zeppelin ID는 ds_handson_20190509로 미리 설정되어 있고, 비밀번호를 재설정하라고 뜨는데

원하는 걸로 임의로 설정해주면 된다.

step5) 데이터셋 준비하기

사용할 데이터는 SKT Big Data Hub에서 제공하는 배달업종 이용현황 분석 2018년 데이터이다.

아래 그림과 같이

S3에 [your_id]-ds-handson-20190509이라는 버켓을 생성하시고 original_data 폴더내에 업로드 한다.

업로드 위치: s3://[your_id]-ds-handson-20190509/original_data/

1) Bucket 생성

aws s3 mb s3://[your_id]-ds-handson-20190509 –region ap-northeast-2

2) 파일 업로드

aws s3 sync s3://data-ds-handson-20190509 s3://[your_id]-ds-handson-20190509/original_data –region ap-northeast-2

3) 업로드 확인

aws s3 ls [your_id]-ds-handson-20190509/original_data/

아래 그림과 같이 aws s3 ls 명령어로 확인하거나 직접 S3 콘솔에서 파일이 업로드 되어있으면 준비가 완료 된 것이다.

2. Zeppelin을 이용한 데이터 전처리

step 1) 제플린 접속하기

브라우저 창하나를 띄우고 아래와 같이 ‘[클러스터의 DNS]:8890’ 를 입력하여 제플린에 접속한다.

그 다음에 화면 우측 상단에 로그인을 클릭한다.

아까 쉘스크립트 실행시 아이디와 임의로 본인이 설정한 비번을 입력하고 로그인한다.

아래 그림과 같이 임의로 노트북을 생성해준다.

12-2

step2) 데이터 불러오기 & 살펴보기

자주쓰는 라이브러리 임포트하고, S3 데이터를 파이스파크를 이용해 불러온다.

%pyspark

from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.sql.window import Window
import pandas as pd
import numpy as np
from urllib.parse import urlparse, parse_qs, unquote_plus, urlsplit
from datetime import datetime, timedelta
from dateutil.relativedelta import relativedelta
import re
import time
import builtins

dat = spark.read.csv("s3n://[your_id]-ds-handson-20190509/original_data/*", header = True)
z.show(dat)

요일별 분석을 위해 요일 정보가 필요하다.

이 페이지를 참고하면 날짜에서 요일정보를 만들 수 있을 것 같아서 새로운 블록을 만들어서 아래와 같이 시도해본다.

새로 만든 “dayofweek” 열에 예상했던 요일정보가 없고 null 값으로 채워져 있다.

사용한 date_format 관련 문서를 보면 입력값으로 date와 format을 받는다. 우리가 넘겨준 ‘일자’칼럼을 date로 인식하지 못해서 발생한 문제인 것으로 판단된다.

%pyspark

dat = spark.read.csv("s3n://[your_id]-ds-handson-20190509/original_data/*", header = True)\
  .withColumn("dow_number", date_format('일자', 'u'))

z.show(dat)

step2) 데이터 전처리하기

‘일자’ 컬럼을 PySpark가 좋아하는 형태로 변환시킨다.

to_date 함수의 foramt 파라미터에 date 정보의 형태를 알려주면 date 형태로 정보를 올바르게 읽어온다.

그 후에 다시 date_format 함수를 이용하여 요일정보를 가져온다.

%pyspark

dat = spark.read.csv("s3n://[your_id]-ds-handson-20190509/original_data/*", header = True)\
  .withColumn("일자", to_date("일자", "yyyyMMdd"))\
  .withColumn("dow_number", date_format('일자', 'u'))\
  .withColumn("dow_string", date_format('일자', 'E'))
  
z.show(dat)

이와같이 Zeppelin을 통해 데이터 전처리가 가능하고 간단한 분석 및 시각화도 가능합니다.

%pyspark

result = dat\
  .groupBy("dow_number", "dow_string")\
  .agg(sum("통화건수").alias("통화건수"))\
  .sort("dow_number")
  
z.show(result)

아래와 같이 차트로 시각화도 할 수 있다.

위의 코드에 ‘업종’을 추가하여 또 다른 시각화 시도

step3) 전처리 데이터 저장하기

날짜포맷을 바꾸고 요일정보를 추가한 데이터를 다시 S3에 적재한다.

기존에는 [your_id]-ds-handson-20190509 버켓의 origianl_data 폴더에 1년치 데이터가 저장되어 있었다.

크게 문제가 되지 않을 파일 사이즈이지만, 만약 일별/월별로 적재되는 데이터의 사이즈가 크고 오랜기간 동안 데이터가 누적되었다면 그 데이터를 조회하는데만 큰 비용이 발생한다.

효율적인 데이터 관리와 파티션이라는 개념을 향후에 사용하기 위해 폴더트리를 월단위로 변경하여 저장하는 방향으로 개선하면 좋다.

%pyspark

start_date = "2018-01-01"

for mm in range(0, 12):
  dt = datetime.strptime(start_date, '%Y-%m-%d') + relativedelta(months = mm)
  date_yyyy_mm = "{:%Y-%m}".format(dt)
  date_simple = "{:%Y-%m-%d}".format(dt)
  print(date_yyyy_mm)

  result = dat\
      .filter(col("일자").startswith(date_yyyy_mm))\
      .repartition(1)\
      .write\
      .csv("s3n://[your_id]-ds-handson-20190509/result/{}".format(date_yyyy_mm), header=True, mode="overwrite")

위의 코드를 실행후 나의 s3버킷으로 가보면 아래 그림과 같이 데이터가 월별로 폴더링이 적재되어 있는것을 볼 수 있다.

3. AWS Glue를 이용한 데이터 카탈로그 생성

step1)

위에서 했던 데이터 처리작업은 S3에 적재되어있는 데이터를 제가 원하는 형태로 가공해서 S3 다른위치에 다시 적재하는 과정이었다.

S3는 사용성이나 비용측면에서 여러가지 장점을 가지고 있어서 실제로 업무를 하면서도 필요에 따라 많은 종류의 그리고 대용량의 처리를 S3에서 무리 없이 사용하고 있다.

작업이 진행될수록 데이터들이 S3내에서도 산재되어 있고, 또한 모든데이터들이 S3에 있는것이 아니라 RDS, DynamoDB 등등 여러 소스에 존재해서 통합 관리의 필요성이 생기는데 이를 위해 AWS Glue 서비스의 가장 강력한 기능인 크롤러를 통해 자동으로 데이터 카탈로그를 생성해본다.

AWS Glue 콘솔화면으로 이동한다.

그리고 아래 그림처럼 데이터베이스로 이동한 다음에 add database를 클릭한다. 데이터베이스 이름은 임의로 지어준다.

만들어진 데이터베이스 하위메뉴에 테이블 클릭해보면 아무런 테이블이 정의되어 있지 않다고 나온다. 테이블 추가 메뉴로 수동으로 테이블을 입력할 수 있지만, 우리는 크롤러를 이용하여 자동으로 테이블을 생성할 것이다. 왼쪽에 크롤러 메뉴로 이동하여 크롤러 추가 버튼을 눌러 크롤러를 생성한다.