데이터 엔지니어링의 필요성

문제해결을 위한 가설 검증 단계

모든 비즈니스가 동일한 데이터 분석 환경을 갖출 수 없으며 성장 단계에 따라 선택 및 집중해야 하는 분석 환경이 다르다.
유저 경험이 중요한 비즈니스의 경우, 처음부터 데이터 시스템 구축이 성공의 열쇠이다.

이커머스는 마케팅, CRM, 물류 데이터 분석을 통해 전략을 수집한다.
처음부터 모든 인력을 갖출 필요는 없고 성장 단계별로 필요한 분석 환경을 갖추는 것이 열쇠이다.

데이터 아키텍쳐 시 고려사항

비즈니스 모델 상 가장 중요한 데이터는 비용 대비 비즈니스 임팩트가 가장 높은 데이터이다.

데이터 거버넌스(Governance)

원칙(Principle)

데이터를 유지 및 관리하기 위한 가이드
보안, 품질, 변경 관리

조직(Organization)

데이터를 관리할 조직의 역할과 책임
데이터 관리자, 데이터 아키텍트

프로세스

데이터 관리를 위한 시스템
작업 절차, 모니터 및 측정

유연하고 변화 가능한 환경 구축

특정 기술 및 솔루션에 얽매여져 있지 않고 새로운 테크를 빠르게 적용할 수 있는 아키텍쳐를 만드는 것이 중요하다.
생성되는 데이터의 형식이 변화할 수 있는 것처럼 그에 맞는 툴과 솔루션도 빠르게 변화할 수 있는 시스템을 구축하는 것이 중요하다.

실시간 데이터 핸들링이 가능한 시스템

밀리세컨 단위의 스트리밍 데이터든, 하루에 한 번 업데이트 되는 데이터든 데이터 아키텍쳐는 모든 스피드의 데이터를 핸들링 해야한다.
- Real Time Streaming 데이터 프로세싱
- Cron이 필요하다.
- Serverles Triggered 데이터 프로세싱

트리을 구축한다. -위험 요소를 파악하여 데이터를 안전하게 트리거## 셀프 서비스 환경을 구축한다.

아티스트 엑세스할 수 있는 프로세스 확장성이 없는 데이터이다.
- 아티스트 Tools
- 프로세스 System for Analysis
- Front-end 데이터 APP

데이터 시스템의 옵션

API

다양한 플랫폼 및 소프트웨어는 API를 통해 데이터를 주고 받을 수 있는 환경을 구축하여 생태계를 생성한다.

RDB

관계형 데이터베이스

NoSQL

하둡/스파크/프레스토(Hadoop/Spark/Presto)

DSS(Distributed Storage System) 및 MapReduce를 통한 병렬 처리가 가능하다.
스파크(Spark)
- 하둡(Hadoop)의 진화 버전이다.
- 리얼 타임 데이터를 프로세싱하기에 최적화 되어 있다.
- 파이썬(Python)을 통한 API를 제공하여 앱을 생성한다.
- SQL 쿼리(Query) 환경을 서포트한다.

서버리스 프레임워크(Serverless Framework)

Triggered by http requests, db events, queuing services
Pay as you use
Form of functions
3rd party APP 및 다양한 API를 통해 데이터를 수집 정제하는데 유용하다.

데이터 파이프라인

데이터를 한 장소에서 다른 장소로 옮긴다.
- API ⇨ DB
- DB ⇨ DB
- DB ⇨ BI

필요한 경우

다양한 데이터 소스로부터 많은 데이터를 생성하고 저장하는 서비스
데이터 사일로(silos): 마케팅, 어카운팅, 세일즈, 오퍼레이션 등 각 영역의 데이터가 서로 고립되어 있는 경우
실시간 혹은 높은 수준의 데이터 분석이 필요한 비즈니스 모델
클라우드 환경이 필요할 경우

구축시 고려사항

확장성(Scalability): 데이터가 기하급수적으로 늘어났을 때도 작동하는가?
안정성(Stability): 에러, 데이터 플로우 등 다양한 모니터링 및 관리가 가능한가?
보안성(Security): 데이터 이동 간 보안에 대한 리스크는 무엇인가?

자동화

데이터를 추출, 수집, 정제하는 프로세싱을 최소한의 사람 인풋으로 머신이 운영하는 것이다.

고려사항

데이터 프로세싱 절차

에러 핸들링 및 모니터링

트리거(Trigger)/스케쥴링(Scheduling)

End-to-End 아키텍쳐

데이터 레이크

예시

넷플릭스

우버

스포티파이 프로젝트 데이터 아키텍쳐

Ad hoc VS Automated

Ad hoc 분석 환경 구축은 서비스를 지속적으로 빠르게 변화시키기 위해 필수적인 요소이다.
이니셜 데이터 삽입, 데이터 백필(Backfill) 등을 위해 Ad hoc 데이터 프로세싱 시스템 구축이 필요하다.
Automated: 이벤트, 스케쥴 등 트리거를 통해 자동화 시스템을 구축한다.

아티스트 관련 데이터 수집 프로세스

데이터 분석 환경 구축

서비스 관련 데이터 프로세스

API

Authentication VS Authorization

Authentication: 정체가 맞다는 것을 증명한다.
Authorization: 어떠한 액션을 허용한다.

OAuth 2.0

Spotify

API를 이용하여 테스트해본다.

import sys
import requests
import base64
import json
import logging

client_id = ""
client_secret = ""


def main():
    headers = get_headers(client_id, client_secret)
    
    # 스포티파이 탐색 API
    params = {
        "q": "BTS",
        "type": "artist",
        "limit": "5"
    }

    r = requests.get("https://api.spotify.com/v1/search", params=params, headers=headers)

    try:
        r = requests.get("https://api.spotify.com/v1/search", params=params, headers=headers)
    except:
        logging.error(r.text)
        sys.exit(1)


    r = requests.get("https://api.spotify.com/v1/search", params=params, headers=headers)

    if r.status_code != 200:
        logging.error(r.text)
        if r.status_code == 429:
            retry_after = json.loads(r.headers)['Retry-After']
            time.sleep(int(retry_after))

            r = requests.get("https://api.spotify.com/v1/search", params=params, headers=headers)
        ## access_token expired
        elif r.status_code == 401:
            headers = get_headers(client_id, client_secret)
            r = requests.get("https://api.spotify.com/v1/search", params=params, headers=headers)
        else:
            sys.exit(1)

    # Get BTS' Albums

    r = requests.get("https://api.spotify.com/v1/artists/3Nrfpe0tUJi4K4DXYWgMUX/albums", headers=headers)

    raw = json.loads(r.text)

    total = raw['total']
    offset = raw['offset']
    limit = raw['limit']
    next = raw['next']

    albums = []
    albums.extend(raw['items'])

    ## 100개만 뽑아온다.
    while next: 
        r = requests.get(raw['next'], headers=headers)
        raw = json.loads(r.text)
        next = raw['next']
        print(next)

        albums.extend(raw['items'])
        count = len(albums)

    print(len(albums))

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, 데이터=payload, headers=headers)

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

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

    return headers

if __name__=='__main__':
    main()