data engineering (데이터 파이프라인 자동화)

데이터 워크 플로우

• 이전에도 언급했었듯이 데이터 파이프라인은 아래와 같은 서비스들을 S3에 모아 Athena같은 서비스로 분석해준 뒤 그 결과를 저장해놓은 일련의 데이터 작업의 흐름을 일컫는다.

• 하나의 job이 시작되거나 어떠한 event에 trigger가 됬을때, 또 다른 job으로 연결이 되는 이런 정보들을 DAGs(Directed Acyclic Graphs)라고 부른다.

ETL

• 보통은 Extract -> Transform -> Load순으로 작업을 해 왔지만, 최근에는 Extract -> Load -> Transform 순으로 작업을 하기도 한다. 데이터 파이프 라인의 연장선이다. 하나의 예시를 들자면, 하루의 정해진 시간에 의한 스케쥴링인 Amazon CloudWatch Event와 Automation Script를 통해서 machine이 시작하면, AWS안에 AWS Step Functions는 각 과정에서 그 다음과정으로의 연결에 대한 여러가지 경우의 수에 대한 룰을 정해놓는 서비스로 쉽게 말하면, 임의의 단계에서 fail이 일어나면 어떤 event를 발생시켜야 하고, success를 하면 어떤 event를 발생시켜야 하는지를 관리할 수 있도록 도와주는 서비스이다. 이런 Step Function안의 ETL Flow state machine이 시작하고, 이후에는 다양한 job들이 작동하게 된다. 이러한 ETL job들의 log를 CloudWatch에 저장을 하고, 아래와 같은 Flow를 갖게된다.

• AWS의 Step function에 관해 조금 더 말하자면, 아래 그림과 같이 사용할 수 있다. start가 되면 job이 submit이 되고, job이 finish될때 까지 기다려 줄 수 있게끔 Wait Seconds를 사용할 수도 있다. 왜냐하면, 예를 들어 Athena는 어느 정도 빅데이터를 처리하는 시스템이기 때문에 MySQL이나 PostgreSQL보다는 느린 부분이 있다. 이런 경우 위와 같이 time sleep을 통해 python script를 잠깐 멈춰두고 그 다음에 해당 시간이 지났을때 그 query에 대한 결과들을 가져올 수 있다. 이후에는 다시 job status를 받고 job이 끝났는지 아닌지에 따라 작업을 진행하는 flow를 볼 수 있다. 이런 service들이 없었을 때는 하나하나 monitoring을 통해서 수동으로 관리를 해야 했다.

• AWS Glue가 가장 좋은 부분은 이전에는 MySQL 같은 경우에는 만들어 놓은 Schema에 맞춰서 data를 insert하였는데, 이제는 data가 너무나 방대해지고 형식도 다 다른데, 이런 것들을 통합하는는 다 Glue한다는 의미의 서비스라는 점이다. 가장 많이 쓰여지는 부분 중에 하나가 Crawler인데 Crawler를 사용하면 자동으로 해당 data를 Crwaling해서 data가 어떤 형식인지에 대해서 지속적으로 Schema 관리가 들어가는 부분이 있다. 그러므로 data 양도 너무나 많고 column도 너무나 많은데 column이 변하는 경우도 있을 경우에 사용하면 좋다.

• AWS Glue 페이지를 보면 아래 그림과 같이 table과 ETL, Trigger등 다양한 작업을 할 수 있다. 한 가지 예시로 S3에 저장해놓은 python Script를 Jobs 탭에서 바로 수정가능하며, Trigger들도 등록해서 관리 할 수 있다.

• 해당 job들은 step function이나 Glue를 통해 관리를 하거나, EC2에서 Crontab으로 스케쥴링의 변화를 통해서 관리를 하는 등 다양한 방법으로 관리를 하지만 아래와 같이 서비스들의 지속적인 monitoring을 통해 cost를 효율적으로 사용할 선택과 집중을 해야 할 것이다. 어떤 부분까지 monitoring을 할 것인지에 대해 선택하여 집중하는 것이다.

Crontab이란?

• 여러가지 만들어진 data job들을 우리가 작성한 해당 코드를 특정한 시간이나 하루에 한번 일주일에 한번이 됐건 어떠한 스케쥴링 방법을 통해서 지속적으로 작동시키려고 할때 사용한다. Crontab Quick reference

• 이러한 스케쥴링을 하려면 규칙에의한 명령이 존재할 것임을 눈치챘을 것이다. 아래 그림과 같이 모두 ‘*‘이면 계속 1분마다 작동하라는 의미이며, 순서대로 분, 시간, 일, 월, 요일 순으로 지정할 수 있다.

• 한가지 간단한 예로는 아래와 같이 Crontab을 실행하면 매일 오후 6시 30분에 /home/someuser/tmp안의 모든 파일을 제거하라는 job을 스케쥴링할 수 있다.

30 18 * * * rm /home/someuser/tmp/*

• 이제 Crontab을 실습해 보기 위해 EC2 서버를 하나 AWS에서 생성해 볼 것이다. 먼저 service 탭에서 EC2를 클릭하여 아래 그림과 같이 EC2 service 페이지로 이동한 후 다시 instance 탭으로 이동한다. Launch instance 버튼을 클릭하면 생성할 instance의 환경을 설정하는 페이지로 이동하게된다.

• Crontab 실습을 위한 서버이기 때문에 사양이 좋은 것을 고르진 않을 것이다. 필자는 Free tier만 사용가능한 2번째 사양을 고를 것이다. 다만, 실무에서 사용할 경우는 각자의 상황에 맞는 서버의 크기와 성능을 골라서 사용해야 할 것이다.

• 필자는 Free Tier만 가능한 t2.micro type을 선택하였다.

• Configure instance 탭에서는 개수를 지정하는 등 instance에 대한 설정을 하는 탭인데, 필자는 default값을 사용하기로 생각해서 아무런 설정도 하지않고 넘어갔다.

• Add storage 탭은 말 그대로 저장 성능을 설정하는 탭으로 필자는 이부분도 별다른 설정없이 기본값으로 하고 넘어갔다.

• tag를 설정하는 탭이며, 필자는 생략하였다.

• Configure Security Group 탭은 Security Group의 설정에 관한 탭이며, 새로 만들수도 있고, 이전에 생성되어있는 그룹을 사용해도 무방하다. 허나, 동일한 inbound 규칙을 사용하지 않는다면 새롭게 생성해서 사용하는 것이 좋다. 필자는 새롭게 만들어서 사용할 것이다. 또한, 접근을 ssh로 할 것이므로 아래와 같이 설정한다.

• 마지막으로 Review 탭은 이제까지 설정한 모든 사항을 점검하고 마지막으로 접속시 사용할 key pair를 어떤것으로 할지 정해주고 나면 모든 과정이 끝이 난다.

• 이제 다시 EC2의 instance 탭을 살펴보면, 새롭게 EC2 서버가 생성된 것을 확인 할 수 있다. 이제 접속을 해볼 것인데, 아래 그림에서 처럼 자신의 pem 파일이 존재하는 path에서 public DNS 서버 주소를 같이 입력하고 접속하면 된다.

• 아래와 같이 command를 실행하면 계속진행할 것이냐는 물음이 나올텐데 yes라고 하면 된다. 이는 앞으로 계속해서 접속하기위해 이 DNS를 추가할 것이냐는 물음이다.

ssh -i pem_file_name.pem ec2-user@Public_DNS

• 이제 Crontab을 실행할 script 파일을 정해야 하는데, 필자는 이전에 만들어 두었던 script 파일 중에 top_tracks와 audio feature들에 대한 데이터를 S3에 parquet화하여 저장하게끔 코드를 작성한 파일을 사용할 것이다. 아래와 같이 필자의 pem 파일은 data_engineering이라는 파일 안에 존재한다.

data_engineering
├── Code
│   ├── Chatbot\ Project
│   │   ├── deploy.sh
│   │   ├── fb_bot.py
│   │   ├── lambda_handler.py
│   │   └── requirements.txt
│   ├── artist_list.csv
│   ├── audio_features.parquet
│   ├── create_artist_genres_table.py
│   ├── create_artist_table.py
│   ├── dynamodb_insert.py
│   ├── select_dynamodb.py
│   ├── spotify_S3.py
│   ├── spotify_s3_artist.py
│   ├── spotify_s3_make.py
│   ├── top-tracks.parquet
│   
├── Slide
├── foxyproxy-settings.xml
└── spotift_chatbot.pem

• 위와 같은 구조로 되어있으므로 사용할 spotify_s3_make.py을 생성한 EC2 Server로 옮겨 줄 것이다. scp은 서버로 파일을 copy하는 것이라고 생각하면 된다. 단,아래 명령어는 이미 EC2에 접속한 상태가 아닌 로컬에서 진행하여야한다. 아래와 같이 옮겨진것을 볼 수 있고, EC2 서버로 접속하여 확인가능하다.

scp -i spotift_chatbot.pem ./Code/spotify_s3_make.py ec2-user@ec2-54-180-97-88.ap-northeast-2.compute.amazonaws.com:~/
spotify_s3_make.py                            100% 5411   536.2KB/s   00:00

ssh -i spotift_chatbot.pem ec2-user@ec2-54-180-97-88.ap-northeast-2.compute.amazonaws.com
ls

• 만일 EC2 서버의 어떤 폴더를 만들어 놓았는데 해당 폴더안으로 이동시키고 싶다면 아래와 같이 하면된다.

scp -i spotift_chatbot.pem ./Code/spotify_s3_make.py ec2-user@ec2-54-180-97-88.ap-northeast-2.compute.amazonaws.com:~/파일이름

• 이제 script를 실행하기에 앞서서 작동할 수 있게끔 먼저 환경을 만들어주어야 할 것이다. 가장 먼저 python3가 설치되어있는지를 확인해 보자.

sudo yum list | grep python3
# 위의 명령문을 실행한 후 python3x에 관한 리스트가 나온다면 python3가 설치되어있는 상태이다.
# 허나 잘 모르겠다면 아래 명령어를 통해 설치하도록하자.
sudo yum install python36 -y

# 그 다음은 pip를 설치해야한다
curl -O https://bootstrap.pypa.io/get-pip.py

sudo python3 get-pip.py

#이제 script파일을 run해본 후에 필요한 패키지들을 추가로 설치해준다.
python3 spotify_s3_make.py

pip install boto3 --user
pip install requests --user
pip install pymysql --user
pip install pandas --user
pip install jsonpath --user

• 이제 본격적으로 crontab을 실습해 볼 것이다. 보통 crontab은 아래 두 가지를 많이 사용한다. crontab 파일은 vim editor를 사용하는데, 해당 스케쥴을 시작할때 email을 보내주는 기능도 있다. 우선 작동시키고 싶은 파일과 파이썬의 위치를 알고있어야 한다.

• UTC 시간 변경

pwd

# 결과
/home/ec2-user

which python3

# 결과
/usr/bin/python3

sudo service crond start

crontab -l

# crontab 파일 쓰는 vim이 켜진 후
# 매일 18시 30분에
30 18 * * * /usr/bin/python3 /home/ec2-user/spotify_s3_make.py

# vim으로 저장한 후 나오면 아래와 같은 메세지가 나와야 정상적으로 crontab을 설정한 것이다.
crontab: installing new crontab

# 어떠한 사항을 crontab으로 스케쥴링하고 있는지 확인하기
crontab -l

# 서버는 UTC를 사용하기 때문에 date 명령어를 통해 현재 서버가 몇시인지를 확인하고 우리나라 시간과 맞추도록 crontab을 설정해 주어야 한다.
date

마이크로서비스에 대한 이해

• 전체적인 수집 프로세스에 대한 설명을 다시 한번 짚고가자면, Unknown Artist가 챗봇 메세지로 들어왔을 경우 AWS Serverless Lamda 서비스를 통해 Spotify API에 Access를 하고 그리고 해당 데이터가 Top Tracks, Artist Table에 가야되는지 또는 S3에 가야되는지를 관리를 하게 될 것이다. 또한, Ad Hoc Data Job을 통해 하루에 한번이라던지, 직접 로컬에서 command line을 통해 데이터를 가져올 수도 있게된다. Lambda가 필요한 이유는 우리가 Unknown Artist가 챗봇 메세지로 들어왔을때 내용을 업데이트를 해야되는데, 보통 사람들이 기대하는 챗봇은 업데이트를 바로 해주어서 원하는 정보를 얻을 수 있게끔 해주어야 하기에 이렇게 바로 업데이트를 할 수 있게끔 Lambda라는 서비스를 통해서 해결 할 수 있다. 이런 Lambda는 마이크로서비스의 개념인데, monolithic 이라는 개념의 반대이다. monolithic은 하나의 서비스를 만들때 크게 프로젝트 단위로 만들어 놓고 관리를 해주는 개념으로써 관리에 있어서 전체 프로세스가 보이기 때문에 컨트롤하기 쉬운 부분이 있다. 이에 반해 마이크로서비스는 세세한 작업하나씩을 단위로 관리를 하는 것이다.

• 챗봇을 Lambda로 구현하는 이유는 Serverless는 하나의 Function이기 때문에 Stateless라고도 하는데 지금 상태가 어떤지 모르겠다는 의미이다. 예를들어, 어떠한 User 어떤 메시지를 보냈다고 가정하면, Lambda Function에는 이전의 어떠한 메시지를 갖고 있었는지를 담고있을 수 없다. 상태가 없는 Function이라고 생각하면 될 것 같다. 그러므로 이런 State를 관리할만한 데이터베이스가 필요할 것이다. 주로 메시지에 특화된 DynamoDB를 사용할 것이다. 또한 Lambda의 경우에는 해당 서비스의 User가 기하급수적으로 늘어났을 때 병렬로 늘어나기 때문에 제한점이 서버로 구현하는것보단 덜하다는 장점도 있다. 서버의 경우에는 메모리나 CPU의 제한된 성능으로 구축된 동일한 서버를 통해 1명에게 서비스하는 것과 백만명에게 서비스하는 것은 완전 다를것이다. 또 한가지 좋은 점은 지속적으로 띄워져 있는 것이 아니라 필요할 때 띄워서 사용한 만큼만 비용을 지불한다는 점이다.

• Crontab을 통하여 Lambda를 호출할 수도 있고, Lambda가 Lambda를 호출할 수도 있다.

• 본격적으로 Lambda Function을 만들 것이다. 먼저 AWS Lambda Function 페이지로 이동하여 아래와 같이 Create Function 버튼을 클릭한다.

• 이번 Lambda Function은 이전에 DynamoDB에 top track정보를 DynamoDB에 저장했었는데, Artist가 추가된다면 DynamoDB에도 저장되어야하므로 이 작업을 작성해 볼 것이다. 아래 그림에서와 같이 Author from scratch는 기본적인 예제를 통해 시작하는 부분이고, Use a blueprint는 흔하게 사용되는 경우들을 코드로 제공하는 부분이다. 마지막은 App repository에서 바로 연결해서 사용하는 것이다. 필자는 Scratch로 진행할 것이다. Lambda는 하나의 Functiond이므로 제한점도 있을 것이다.

• 아래 그림과 같이 Function의 이름을 정하고 function의 language를 정한다. 또한 가장 중요한 부분인 Permission부분이 남았는데, 이 부분은 현재 필자가 진행할 Function의 목표는 DynamoDB에 새로운 데이터를 추가하는 것이므로 DynamoDB에 대한 permission을 갖고 있어야 오류가 없을 것이다. 그렇기에 2번째 부분인 Use an exsiting role을 클릭하여 사용해야 하는데, 새롭게 규칙을 추가해서 사용하면 error가 어떻게 발생하는 지를 보기위해 우선 첫번째를 선택하였다. 물론 무조건적으로 있던 규칙을 사용하는 것이 아니라 상황에 맞춰 사용해야한다.

• function을 생성하면 아래 그림과 같이 여러가지 설정 및 작업을 할 수 있는 페이지가 나온다.

• 아래 부분으로 내려보면 다음과 같이 Function의 코드를 작성할 수 있는 부분이 존재한다. 이 곳에서 Function을 정의할 것이다. 허나, Edit code inline은 거의 사용하지 못하는 설정이라고 볼 수 있다. 해당 Lambda Function은 Linux 기반의 AMI compute system에 있는데, 함수를 동작할 수 있는 패키지들이 아무런 설치나 설정이 되어있지 않은 상태이기 때문이다. 그래서 zip file을 업로드하거나 S3에서 불러오는 방식을 보통 채택한다. 필자는 S3에서 불러오는 방식을 사용할 것이다.

• 위에서 함수에 사용될 변수들을 정의할 수 있다. 예를 들어, Spotify API에 접속하기 위해서는 ID와 Secret Key가 필요했는데 이 부분을 코드에 적기 보다는 보안의 문제로 따로 변수로 처리해 둔 뒤 함수에는 그 값을 받아 사용하는 형식으로 사용된다. 예를 들면 아래 그림에서 환경 변수의 Edit을 클릭하면 변수의 Key와 Value를 입력하여 추가하게끔 되어있는데 추가했다고 가정하면 함수에서는 os.environ.get(key)로 값을 받으면 된다. 또한 Basic settings부분은 Memory(Max : 3GB)와 timeout(Max : 15분)을 설정할 수 있는 부분인데, 가장 간단하고 명료하게 병렬적으로 분산처리를 할 수 있도록 코드를 작성하는 것이 좋다는 Lambda의 특징을 잘 보여주는 부분이다. 여기서 Memory는 크게 해 놓아도 사용한 만큼만 비용을 지불하는 것이므로 상관없다는 것에 유의하자.

• 이미 해당 Artist가 존재하는 것은 확인한 상태여서 해당 Artist ID에 대한 top track만 확보하고 싶은 경우라고 가정할 것이다. 그 ID 값을 이 Lambda Function에 보내 줌으로써 다시 한번 Spotify API에 hit을 하여 top track 정보를 가져오고 다시 DynamoDB에 저장하도록 할 것이다.

• 본격적으로 Lambda Function을 만들어 볼 것이다. 우선 새로운 폴더를 만들어준다. 그 안에는 Lambda Function에 관한 것들만 담기 위해서이다. 위에서 언급 한 것처럼 우선 아래 패키지들을 shell script를 통해 실행하기 위해서 requirements.txt를 작성할 것이다. 헌데 AWS에는 기본적으로 boto3가 설치되어져 있고 나머지들은 기본 내장 패키지들이므로 requests만 설치해 주면 될 것 같다.

• requirements.txt

requests

• 위에서 만들어 놓은 requirements.txt안의 패키지를 -t옵션의 target 파일에 저장한다는 의미이다. 헌데 이렇게 home이 아닌 파일에 저장하게 되면 pointer issue 때문에 error가 발생하는데 이는 새롭게 setup.cfg라는 파일안에 아래와 같이 작성한 후에 저장해주면 해결된다. 다시 아래의 명령문을 실행하면 error 없이 libs안에 설치가 될 것이다.

pip install -r requirements.txt -t ./libs

• setup.cfg

[install]
prefix=

• 매번 우리가 AWS CLI를 통해 명령문을 치고 실행할 수 없으므로 shell script로 작성해 준다.

• deploy.sh

#!/bin/bash

rm -rf ./libs
pip3 install -r requirements.txt -t ./libs

rm *.zip
zip top_tracks.zip -r *

aws s3 rm s3://top-tracks-lambda/top_tracks.zip
aws s3 cp ./top_tracks.zip s3://top-tracks-lambda/top_tracks.zip
aws lambda update-function-code --function-name top-tracks --s3-bucket top-tracks-lambda --s3-key top_tracks.zip

또한, 위에서 lambda function을 update할 s3 bucket을 지정했으므로 새롭게 위의 이름으로 생성한다. 이 단계는 먼저 bucket을 만든뒤에 앞의 shell script를 만들어도 무관하다.

• lambda_function.py

import sys
sys.path.append('./libs')
import os
import boto3
import requests
import base64
import json
import logging


client_id = "Your Spotify Developer ID"
client_secret = "Your Spotify Developer PW"

try:
    dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-2', endpoint_url='http://dynamodb.ap-northeast-2.amazonaws.com')
except:
    logging.error('could not connect to dynamodb')
    sys.exit(1)


def lambda_handler(event, context):

    headers = get_headers(client_id, client_secret)

    table = dynamodb.Table('top_tracks')

    artist_id = event['artist_id']

    URL = "https://api.spotify.com/v1/artists/{}/top-tracks".format(artist_id)
    params = {
        'country': 'US'
    }
    r = requests.get(URL, params=params, headers=headers)

    raw = json.loads(r.text)

    for track in raw['tracks']:

        data = {
            'artist_id': artist_id
        }

        data.update(track)

        table.put_item(
            Item=data
        )
    # AWS CloudWatch에서 Log기록을 살펴볼때 확인하기 위해 return 값을 Success로 주었다.
    return "SUCCESS"


def get_headers(client_id, client_secret):

    endpoint = "https://accounts.spotify.com/api/token"
    encoded = base64.b64encode("{}:{}".format(client_id, client_secret).encode('utf-8')).decode('ascii')

    headers = {
        "Authorization": "Basic {}".format(encoded)
    }

    payload = {
        "grant_type": "client_credentials"
    }

    r = requests.post(endpoint, data=payload, headers=headers)

    access_token = json.loads(r.text)['access_token']

    headers = {
        "Authorization": "Bearer {}".format(access_token)
    }

    return headers


if __name__=='__main__':
    main()

• 위와 같이 파일을 모두 작성했다면 아래와 같은 구조로 만들어져 있을 것이다.

top_tracks
├── deploy.sh
├── lambda_function.py
├── requirements.txt
└── setup.cfg

• 이제 top_tracks의 path에서 아래와 같이 shell script를 실행시켜준다.

./deploy.sh

# 결과는 아래와 같이 permission denied가 발생할 것이다.
-bash: ./deploy.sh: Permission denied

• Permission 권한을 바꿔주기 위해 다음의 코드를 실행한뒤 다시 shell script를 실행시킨다.

  chmod +x deploy.sh
  
  ./deploy.sh

• 위의 코드들을 실행한 뒤 다시 AWS Lambda Function의 페이지로 돌아가 보면 아래와 같이 해당 파일들을 사용할 수 있게 설정이 되어져 있는 것을 확인 할 수 있다. 보통은 아래와 같이 sensitive한 정보들(client_id, client_secret) 같은 정보들은 따로 config.py 파일을 만들어서 모아놓는다. 이런 config.py파일은 해당 관리자만 가지고 있도록 하여 보안에 유지하나, 필자는 혼자 사용하므로 따로 만들지 않았다.

• 만약 위에서 Function Code내에서는 sensitive한 정보들이 안보이도록 하려면 아래 그림과 같이 environment variable을 추가하고, 코드를 작성하면된다. key값과 value값이 문자라도 기호를 사용하지 않고 넣으면 된다.

• 이제 Lambda Function을 Test해 볼 차례이다. Test 버튼을 클릭해 준 뒤, event의 artist_id값을 주기 위해 RDS에 접속해서 임의의 ID 하나를 필자는 가져왔다.

• Test 데이터를 저장하면 아래 그림과 같이 테스트할 파일로 바뀌었다는 것을 확인 할 수 있다. 이제 Test 버튼을 클릭한다. Log를 보니 Error가 발생한 것 같으므로 먼저 확인해 볼 것이다. Log는 AWS CloudWatch 서비스에서 확인할 수 있다.

• 아래 로그 중 빨간색 박스 부분을 살펴보면 해당 role에 대한 permission이 없기 때문에 발생한 error임을 확인 할 수 있다.

• 처음에 Lambda Function을 만들때 말했듯이 permission이 없기 때문에 발생하는 error이므로 excution role을 변경해주기 위해 아래 그림의 빨간색 박스 안의 버튼을 클릭해 준다.

• role을 변경하기 위해 IAM 페이지로 이동되며, Attach를 통해 DynamoDB에 access 할 수 있는 권한을 부여할 것이다.

• 새롭게 role이 추가되어 기존의 1개에서 2개로 늘어났으며, DynamoDBFullAccess 권한을 부여받음을 확인할 수 있다.

• 이제 다시 Lambda Function 페이지로 돌아와서 실행시켜 보면 제대로 실행됨을 알 수 있다.

• Lambda는 Event trigger 뿐만 아니라, Crontab과 같이 스케쥴링에 의한 job을 작동시킬수도 있으며, 적용가능한 여러가지 event들이 존재한다.