[3주차] ETL/데이터파이프라인

비구조화된 데이터 처리하기

• LOG
  대표적으로 데이터레이크에 저장하는 데이터. 원본 상태로 데이터레이크에 넣어둔 상태에서, 이걸 정제하고 가공해서 데이터웨어하우스에 적재한다.
• AMZONE S3(STROAGE)
  • 클라우드 판 하드디스크라고 생각하면 좋다. 1테라 바이트당 한달에 만원 꼴. 온갖 종료의 데이터를 별다른 비용 걱정 없이 저장할 수 있다. 이 아마존 S3 같은 것을 데이터레이크(DATA LAKE)라고 한다.
  • 데이터레이크
    보존기한이 없는 모든 데이터를 원래 형태대로 보존하는 스토리지, 보통은 데이터 웨어하우스에 비해 몇 배 더 크다
• APACHE SPARK, AMAZON ATHENA
  • ATHENA는 BIG DATA SQL로 생각하면 좋다
  • SPARK도 SPARK SQL 혹은 DATAFRAME으로 판다스처럼 사용할 수 있다.
    웬만해서는 SQL을 쓰지만, 복잡한 경우에만 DATAFRAME을 사용한다.
  • SPARK, ATHENA는 데이터레이크에 있는 데이터를 정제하고 크기를 줄인 다음에, REDSHIFT(데이터웨어하우스)에 저장한다.
  • 보통 현업에서는 SQL 형태로 많이 사용하고, 데이터프레임은 할 수록 번거로워서 잘 못쓴다. 또 많은 사람들이 사용할 수 있기 때문에 SQL을 선호

대용량 데이터 병렬 처리(FEATURE 계산)

추천 알고리즘 등을 주기적으로 계산해놓아야하는 경우에 사용한다.

• 배치로 AIRFLOW, SPARK JOB을 사용해서 NOSQL(카산드라)에 저장해놓으면 RUNTIME에 사용자가 로그인할 때, 그 사용자에 관한 FEATURE들을 NOSQL에서 읽어다가 RECOMMENDATION 모델의 인풋으로 줘서, 추천할 것들을 저장해둠
• 특히 SPARK은 SPARK STREAMLING이라는 것도 있는데, 실시간(REAL TIME)으로 할 수도 있다(빅데이터 프로세싱 종합선물세트 느낌)

---

데이터레이크와 데이터웨어하우스

• 데이터레이크
  • 구조화 데이터 + 비구조화 데이터
  • 보존 기한이 없는 모든 데이터를 원래 형태대로 보존하는 스토리지에 가까움
  • 보통은 데이터 웨어하우스보다 몇배는 더 큰 스토리지
• 데이터웨어하우스
  • 보존 기한이 있는 구조화된 데이터를 저장하고 처리하는 스토리지
  • 보통 BI 툴들(룩커, 태블로, 수퍼셋, …)은 데이터 웨어하우스를 백엔드로 사용함

데이터 레이크와 ELT

• 데이터 소스는 전부 데이터 레이크에 저장한다.(대표적으로 AMAZON S3)
• 그 다음 SPARK, ATHENA를 통해서 TRANSFORM 된 것들을 DATA WAREHOUSE, DATA MART에 저장

---

데이터 파이프라인

• 데이터를 소스부터 목적지로 복사하는 작업
  • 이 작업은 보통 코딩(파이썬 혹은 스칼라) 혹은 SQL을 통해 이뤄짐
  • 목적지는 주로 데이터웨어하우스, 외에도 프로덕션 데이터베이스, NOSQL, S3.. 등
• 데이터 파이프라인의 종류
  • Raw data ETL jobs
    내,외부 데이터 소스에서 데이터를 읽어서(많은 경우 API 활용), 적당한 데이터 포맷 변환 후(데이터 크기 커지면 SPARK), 데이터 웨어하우스 로드, 보통 데이터 엔지니어가 함
  • Summary / Report jobs
    • 데이터웨어하우스 혹은 데이터레이크로부터 데이터를 다시 데이터웨어하우스에 쓰는 ETL
    • RAW DATA를 읽어서 일종의 리포트 형태나 써머리 형태의 테이블을 다시 만드는 용도
    • 특수한 형태로는 AB 테스트 결과를 분석하는 데이터 파이프라인도 존재
    • 요약 테이블의 경우 SQL(CTAS)만으로 만들고 이는 데이터분석가 ROLE이 맞음, 어떻게 분석가들이 편하게 할 수 있는 환경을 만드느냐가 관건
  • Production Data Jobs
    • DW로부터 데이터를 읽어 다른 STROAGE(많은 경우 프로덕션 환경)로 쓰는 ETL
      a. 써머리 정보가 프로덕션 환경에서 성능 이유로 필요한 경우
      b. 머신러닝 모델에서 필요한 피쳐들을 미리 계산해두는 경우
    • 이 경우 흔한 타겟 스토리지:
      • cassandra/HBASE/DynamoDB와 같은 NOSQL
      • MYSQL과 같은 관계형 데이터베이스(OLTP)
      • REDIS/MEMCACHE와 같은 캐시
      • ELASTICSEARCH와 같은 검색엔진

---

간단한 ETL 작성

ETL

• Extract:
  데이터를 데이터 소스에서 읽어내는 과정
• Transform:
  필요하다면 그 원본 데이터의 포맷을 원하는 형태로 변경시키는 과정. 굳이 변환할 필요는 없다
• Load:
  최종적으로 Data Warehouse에 테이블로 집어넣는 과정

과정

• PSYCOPG2 활용 레드쉬프트와 연동
• 외부 데이터를 읽어서, 판다스를 활용해서 TRANSFORM 후에
• INSERT를 통해 적재(보통 COPY를 많이 씀)
  이 과정에서 데이터의 중복값 처리와, 멱등성(동일한 요청을 한 번 보내는 것과 여러 번 연속으로 보내는 것이 같은 효과를 지니고, 서버의 상태도 동일하게 남을 때)을 보장해야함

예시

import pandas as pd


def sql_execute(cur, value1, value2):
  ### NOT EXISTS 사용하여 중복될 경우 INSERT 하지 않음
  sql = f''' 
  INSERT INTO {my_table} (name, gender)
  SELECT '{value1}', '{value2}'
  WHERE
  NOT EXISTS (
    SELECT name, gender
    FROM {my_table}
    WHERE  name = '{value1}' AND  gender = '{value2}');
  '''
  print(sql)
  cur.execute(sql)

### PANDAS READ_CSV 사용하여 value만 사용(헤더 값 사용 X)
def transform_load(link, my_table):
  df = pd.read_csv(link)

  # 결측치 제거 및 중복 값 제거
  df = df.dropna().drop_duplicates()

  cur = get_Redshift_connection()
  df.apply(lambda x : sql_execute(cur, x[0], x[1]), axis = 1)