• 데이터 분야에서 일하고자 하면 반드시 익혀야할 기본 기술
• 구조화된 데이터를 다루는한 SQL은 데이터 규모와 상관없이 쓰임
• 모든 대용량 데이터 웨어하우스는 SQL 기반
  • Redshift, Snowflake, BigQuery
  • Hive/Presto
• Spark도 예외는 아님
  • Spark SQL이 지원됨

Spark SQL이란?

• Spark SQL은 구조화된 데이터 처리를 위한 Spark 모듈
• 데이터 프레임 작업을 SQL로 처리 가능
  • 데이터프레임에 테이블 이름 지정 후 sql함수 사용가능
    • 판다스에도 pandasql 모듈의 sqldf 함수를 이용하는 동일한 패턴 존재
  • HQL(Hive Query Language)과 호환 제공
    • Hive 테이블들을 읽고 쓸 수 있음 (Hive Metastore)

Spark SQL vs. DataFrame

• 하지만 SQL로 가능한 작업이라면 DataFrame을 사용할 이유가 없음
  • 두 개를 동시에 사용할 수 있다는 점 분명히 기억

1. Familiarity/Readability
   • SQL이 가독성이 더 좋고 더 많은 사람들이 사용가능
2. Optimization
   • Spark SQL 엔진이 최적화하기 더 좋음 (SQL은 Declarative)
     • Catalyst Optimizer와 Project Tungsten
3. Interoperability/Data Management
   • SQL이 포팅도 쉽고 접근권한 체크도 쉬움

Spark SQL 사용법 - SQL 사용 방법

• 데이터 프레임을 기반으로 테이블 뷰 생성: 테이블이 만들어짐
  • createOrReplaceTempView: spark Session이 살아있는 동안 존재
  • createOrReplaceGlobalTempView: Spark 드라이버가 살아있는 동안 존재
• Spark Session의 sql 함수로 SQL 결과를 데이터 프레임으로 받음

namegender_df.createOrReplaceTempView("namegender")
 namegender_group_df = spark.sql("""
    SELECT gender, count(1) FROM namegender GROUP BY 1
 """)
 print(namegender_group_df.collect())

SparkSession 사용 외부 데이터베이스 연결

• Spark Session의 read 함수를 호출(로그인 관련 정보와 읽어오고자 하는 테이블 혹은 SQL을 지정). 결과가 데이터 프레임으로 리턴됨

df_user_session_channel = spark.read \
   .format("jdbc") \
   .option("driver", "com.amazon.redshift.jdbc42.Driver") \
   .option("url", "jdbc:redshift://HOST:PORT/DB?user=ID&password=PASSWORD") \
   .option("dbtable", "raw_data.user_session_channel") \
   .load()

Aggregation 함수

• DataFrame이 아닌 SQL로 작성하는 것을 추천

• 뒤에서 실습시 예를 몇 가지 SQL로 살펴볼 예정

  • Group By
  • Window
  • Rank

JOIN

• SQL 조인은 두 개 혹은 그 이상의 테이블들을 공통 필드를 가지고 머지
• 스타 스키마로 구성된 테이블들로 분산되어 있던 정보를 통합하는데 사용
• 왼쪽 테이블을 LEFT라고 하고 오른쪽 테이블을 RIGHT이라고 하면
  • JOIN의 결과는 방식에 따라 양쪽의 필드를 모두 가진 새로운 테이블을 생성
  • 조인의 방식에 따라 다음 두 가지가 달라짐
    • 어떤 레코드들이 선택되는지?
    • 어떤 필드들이 채워지는지?

JOIN 실습 - 예제 데이터 준비

INNER JOIN

1. 양쪽 테이블에서 매치가 되는 레코드들만 리턴함
2. 양쪽 테이블의 필드가 모두 채워진 상태로 리턴됨

   SELECT * FROM Vital v
   JOIN Alert a ON v.vitalID = a.vitalID;

LEFT JOIN

1. 왼쪽 테이블(Base)의 모든 레코드들을 리턴함
2. 오른쪽 테이블의 필드는 왼쪽 레코드와 매칭되는 경우에만 채워진 상태로 리턴됨

   SELECT * FROM raw_data.Vital v
   LEFT JOIN raw_data.Alert a ON v.vitalID = a.vitalID;

FULL JOIN

1. 왼쪽 테이블과 오른쪽 테이블의 모든 레코드들을 리턴함
2. 매칭되는 경우에만 양쪽 테이블들의 모든 필드들이 채워진 상태로 리턴됨

   SELECT * FROM raw_data.Vital v
   FULL JOIN raw_data.Alert a ON v.vitalID = a.vitalID;

CROSS JOIN

1. 왼쪽 테이블과 오른쪽 테이블의 모든 레코드들의 조합을 리턴함

   SELECT * FROM raw_data.Vital v CROSS JOIN raw_data.Alert a;

SELF JOIN

1. 동일한 테이블을 alias를 달리해서 자기 자신과 조인함

   SELECT * FROM raw_data.Vital v1
   JOIN raw_data.Vital v2 ON v1.vitalID = v2.vitalID;

최적화 관점에서 본 조인의 종류들

• Shuffle JOIN
  • 일반 조인 방식
  • Bucket JOIN: 조인 키를 바탕으로 새로 파티션을 새로 만들고 조인을 하는 방식
• Broadcast JOIN
  • 큰 데이터와 작은 데이터 간의 조인
  • 데이터 프레임 하나가 충분히 작으면 작은 데이터 프레임을 다른 데이터 프레임이 있는 서버들로 뿌리는 것 (broadcasting)
    • spark.sql.autoBroadcastJoinThreshold 파라미터로 충분히 작은지 여부 결정

UDF란 무엇인가?

• User Defined Function
• DataFrame이나 SQL에서 적용할 수 있는 사용자 정의 함수
• Scalar 함수 vs. Aggregation 함수
  • Scalar 함수 예: UPPER, LOWER, …
  • Aggregation 함수 (UDAF) 예: SUM, MIN, MAX

UDF 사용 방법

• 함수 구현

  • 파이썬 람다 함수
  • 파이썬 (보통) 함수
  • 파이썬 판다스 함수:
    • pyspark.sql.functions.pandas_udf로 annotation
    • Apache Arrow를 사용해서 파이썬 객체를 자바 객체로 변환이 훨씬 더 효율적

• 함수 등록

  • pyspark.sql.functions.udf

    • DataFrame에서만 사용 가능

  • spark.udf.register

    • SQL 모두에서 사용 가능

• 함수 사용

  • .withColumn, .agg
  • SQL

성능이 중요하다면?

Scala나 Java로 구현하는 것이 제일 좋음 파이썬을 사용해야한다면 Pandas UDF로 구현

Spark 데이터베이스와 테이블

• 카탈로그: 테이블과 뷰에 관한 메타 데이터 관리
  • 기본으로 메모리 기반 카탈로그 제공 - 세션이 끝나면 사라짐
  • Hive와 호환되는 카탈로그 제공 - Persistent
• 테이블 관리 방식
  • 테이블들은 데이터베이스라 부르는 폴더와 같은 구조로 관리 (2단계)
• 메모리 기반 테이블/뷰:
  • 임시 테이블로 앞서 사용해봤음
• 스토리지 기반 테이블
  • 기본적으로 HDFS와 Parquet 포맷을 사용
  • Hive와 호환되는 메타스토어 사용
  • 두 종류의 테이블이 존재 (Hive와 동일한 개념)
    • Managed Table

        - Spark이 실제 데이터와 메타 데이터 모두 관리

    • Unmanaged (External) Table

        - Spark이 메타 데이터만 관리

Spark SQL - 스토리지 기반 카탈로그 사용 방법

• Hive와 호환되는 메타스토어 사용
• SparkSession 생성시 enableHiveSupport() 호출
  • 기본으로 “default”라는 이름의 데이터베이스 생성

from pyspark.sql import SparkSession
 spark = SparkSession \
   .builder \
   .appName("Python Spark Hive") \
   .enableHiveSupport() \
   .getOrCreate()

Spark SQL - Managed Table 사용 방법

• 두 가지 테이블 생성방법
  • dataframe.saveAsTable("테이블이름")
  • SQL 문법 사용 (CREATE TABLE, CTAS)
• spark.sql.warehouse.dir가 가리키는 위치에 데이터가 저장됨
  • PARQUET이 기본 데이터 포맷
• 선호하는 테이블 타입
• Spark 테이블로 처리하는 것의 장점 (파일로 저장하는 것과 비교시)
  • JDBC/ODBC등으로 Spark을 연결해서 접근 가능 (태블로, 파워BI)

Spark SQL - External Table 사용 방법

• 이미 HDFS에 존재하는 데이터에 스키마를 정의해서 사용
  • LOCATION이란 프로퍼티 사용
• 메타데이터만 카탈로그에 기록됨
  • 데이터는 이미 존재.
  • External Table은 삭제되어도 데이터는 그대로임

    CREATE TABLE table_name (
    column1 type1,
    column2 type2,
    column3 type3,
    …
    )
    USING PARQUET
    LOCATION 'hdfs_path';

    유닛 테스트란?

• 코드 상의 특정 기능 (보통 메소드의 형태)을 테스트하기 위해 작성된 코드
• 보통 정해진 입력을 주고 예상된 출력이 나오는지 형태로 테스트
• CI/CD를 사용하려면 전체 코드의 테스트 커버러지가 굉장히 중요해짐
• 각 언어별로 정해진 테스트 프레임웍을 사용하는 것이 일반적
  • JUnit for Java
  • NUnit for .NET
  • unittest for Python

• “서버 한대로 처리할 수 없는 규모의 데이터”
• 2012년 4월 아마존 클라우드 컨퍼런스에서 아마존의 data scientist인 존 라우저(John Rauser)가 내린 정의 분산 환경이 필요하느냐에 포커스
• “기존의 소프트웨어로는 처리할 수 없는 규모의 데이터”
• 대표적인 기존 소프트웨어 오라클이나 MySQL과 같은 관계형 데이터베이스
  • 분산환경을 염두에 두지 않음
  • Scale-up 접근방식 (vs. Scale-out)
    • 메모리 추가, CPU 추가, 디스크 추가

4V (Volume, Velocity, Variety, Varecity)

• Volume: 데이터의 크기가 대용량?
• Velocity: 데이터의 처리 속도가 중요?
• Variety: 구조화/비구조화 데이터 둘다?
• Veracity: 데이터의 품질이 좋은지?

빅데이터 처리의 특징은?

• 먼저 큰 데이터를 손실없이 보관할 방법이 필요: 스토리지
• 처리 시간이 오래 걸림: 병렬처리
• 이런 데이터들은 비구조화된 데이터일 가능성이 높음: SQL만으로는 부족
  • 예를 들면 웹 로그 파일

해결 방안은?

• 큰 데이터를 손실없이 보관할 방법이 필요
  • 큰 데이터 저장이 가능한 분산 파일 시스템이 필요
• 시간이 오래 걸림
  • 병렬 처리가 가능한 분산 컴퓨팅 시스템이 필요
• 이런 데이터들은 비구조화된 데이터일 가능성이 높음
  • 비구조화 데이터를 처리할 방법이 필요

대용량 분산 시스템이란?

• 분산 환경 기반 (1대 혹은 그 이상의 서버로 구성)
  • 분산 파일 시스템과 분산 컴퓨팅 시스템이 필요
• Fault Tolerance
  • 소수의 서버가 고장나도 동작해야함
• 확장이 용이해야함
  • Scale Out이 되어야함

하둡(Hadoop)의 등장

• Doug Cutting이 구글랩 발표 논문들에 기반해 만든 오픈소스 프로젝트
  • 2003년 The Google File System
  • 2004년 MapReduce: Simplified Data Processing on Large Cluster
• 처음 시작은 Nutch라는 오픈소스 검색엔진의 하부 프로젝트
  • 하둡은 Doug Cutting의 아들의 코끼리 인형의 이름
  • 2006년에 아파치 톱레벨 별개 프로젝트로 떨어져나옴

하둡(Hadoop)이란?

• Hortonworks의 정의
  • An open source software platform for distributed storage and distributed processing of very large data sets on computer clusters built from commodity hardware
• 다수의 노드로 구성된 클러스터 시스템 (Cluster)
  • 마치 하나의 거대한 컴퓨터처럼 동작
  • 사실은 다수의 컴퓨터들이 복잡한 소프트웨어로 통제됨

하둡(Hadoop)의 발전

• 하둡 1.0은 HDFS위에 MapReduce라는 분산컴퓨팅 시스템이 도는 구조
  • MapReduce 위에서 다양한 컴퓨팅 언어들이 만들어짐
• 하둡 2.0에서 아키텍처가 크게 변경됨
  • 하둡은 YARN이란 이름의 분산처리 시스템위에서 동작하는 애플리케이션이 됨
  • Spark은 YARN위에서 애플리케이션 레이어로 실행됨

HDFS - 분산 파일 시스템

• 데이터를 블록단위로 나눠 저장
  • 블록의 크기는 128 MB (디폴트)
• 블록 복제 방식 (Replication)
  • 각 블록은 3 군데에 중복 저장됨
  • Fault tolerance를 보장할 수 있는 방식으로 이 블록들은 저장됨
• 하둡 2.0 내임노드 이중화 지원
  • Active & Standby
    • 둘 사이에 share edit log가 존재
  • Secondary 내임노드는 여전히 존재

MapReduce: 분산 컴퓨팅 시스템

• 하둡 1.0
• 하나의 잡 트래커와 다수의 태스크 트래커로 구성됨
  • 잡 트래커가 일을 나눠서 다수의 태스크 트래커에게 분배
  • 태스크 트래커에서 병렬처리
• MapReduce만 지원
  • 제너럴한 시스템이 아님

분산 컴퓨팅 시스템: 하둡 2.0 (YARN 1.0)

• 세부 리소스 관리가 가능한 범용 컴퓨팅 프레임웍
  • 리소스 매니저
    • Job Scheduler, Application Manager
  • 노드 매니저
  • 컨테이너
    • 앱 마스터
    • 태스크
• Spark이 이 위에서 구현됨

  YARN의 동작

1. 실행 코드(와 환경 정보)를 RM에게 제출
2. RM이 NM을 통해 AM 실행
3. AM이 RM으로 코드에 실행에 필요한 리소스를 받아옴
4. AM이 NM을 통해 컨테이너들을 받아 코드 실행 (태스크)
5. 태스크들은 자신의 상황을 주기적으로 AM에게 업데이트 (heartbeat)

• 실행하려는 코드와 환경 정보를 RM(Resource Manager)에게 넘김 i. 실행에 필요한 파일들은 application ID에 해당하는 HDFS 폴더에 미리 복사됨

• RM은 NM(Node Manager)으로부터 컨테이너를 받아 AM(Application Master) 실행 i. AM은 프로그램 마다 하나씩 할당되는 프로그램 마스터에 해당

• AM은 입력 데이터 처리에 필요한 리소스를 RM에게 요구 i. RM은 data locality를 고려해서 리소스(컨테이너)를 할당

• AM은 할당받은 리소스를 NM을 통해 컨테이너로 론치하고 그 안에서 코드를 실행 i. 이 때 실행에 필요한 파일들이 HDFS에서 Container가 있는 서버로 먼저 복사

• 각 태스크는 상황을 주기적으로 AM에게 보고 (heartbeat) i. 태스크가 실패하거나 보고가 오랜 시간 없으면 태스크를 다른 컨테이너로 재실행

하둡 1.0 vs. 하둡 2.0

• 하둡 2.0에서 소개된 클러스터 자원 관리자를 YARN이라고 부름

하둡 3.0의 특징

• YARN 2.0을 사용
  • YARN 프로그램들의 논리적인 그룹(플로우라고 부름)으로 나눠서 자원 관리가 가능. 이를 통해 데이터 수집 프로세스와 데이터 서빙 프로세스를 나눠서 관리 가능
  • 타임라인 서버에서 HBase를 기본 스토리지로 사용 (하둡 2.1)
• 파일 시스템
  • 내임노드의 경우 다수의 스탠바이 내임노드를 지원
  • HDFS, S3, Azure Storage 이외에도 Azure Data Lake Storage 등을 지원

맵리듀스 프로그래밍의 특징

• 데이터 셋은 Key, Value의 집합이며 변경 불가(immutable)
• 데이터 조작은 map과 reduce 두 개의 오퍼레이션으로만 가능
  • 이 두 오퍼레이션은 항상 하나의 쌍으로 연속으로 실행됨
  • 이 두 오퍼레이션의 코드를 개발자가 채워야함
• 맵리듀스 시스템이 Map의 결과를 Reduce단으로 모아줌
  • 이 단계를 보통 셔플링이라 부르며 네트웍단을 통한 데이터 이동이 생김

맵리듀스 프로그래밍의 핵심: 맵과 리듀스

• Map: (k, v) -> [(k', v')*]

  • 입력은 시스템에 의해 주어지며 입력으로 지정된 HDFS 파일에서 넘어옴
  • 키,밸류 페어를 새로운 키,밸류 페어 리스트로 변환 (transformation)
  • 출력: 입력과 동일한 키, 밸류 페어를 그대로 출력해도 되고 출력이 없어도 됨

• Reduce: (k’, [v1’, v2’, v3’, v4’, …]) -> (k’’, v'')

  • 입력은 시스템에 의해 주어짐
  • 맵의 출력 중 같은 키를 갖는 키/밸류 페어를 시스템이 묶어서 입력으로 넣어줌
  • 키와 밸류 리스트를 새로운 키,밸류 페어로 변환
  • SQL의 GROUP BY와 흡사
  • 출력이 HDFS에 저장됨

MapReduce 프로그램 동작 예시

• Word Count

MapReduce: Shuffling and Sorting

• Shuffling
  • Mapper의 출력을 Reducer로 보내주는 프로세스를 말함
• Sorting
  • 전송되는 데이터의 크기가 크면 네트웍 병목을 초래하고 시간이 오래 걸됨
  • 모든 Mapper의 출력을 Reducer가 받으면 이를 키별로 소팅

MapReduce: Data Skew

• 각 태스크가 처리하는 데이터 크기에 불균형이 존재한다면?
  • 병렬처리의 큰 의미가 없음. 가장 느린 태스크가 전체 처리 속도를 결정
  • 특히 Reducer로 오는 데이터 크기는 큰 차이가 있을 수 있음
    • Group By나 Join등에 이에 해당함
    • 처리 방식에 따라 Reducer의 수에 따라 메모리 에러등이 날 수 있음
      • 데이터 엔지니어가 고생하는 이유 중의 하나
      • 빅데이터 시스템에는 이 문제가 모두 존재

MapReduce 프로그래밍의 문제점

• 낮은 생산성
  • 프로그래밍 모델이 가진 융통성 부족 (2가지 오퍼레이션만 지원)
  • 튜닝/최적화가 쉽지 않음
    • 예) 데이터 분포가 균등하지 않은 경우
• 배치작업 중심
  • 기본적으로 Low Latency가 아니라 Throughput에 초점이 맞춰짐

MapReduce 대안들의 등장

• 더 범용적인 대용량 데이터 처리 프레임웍들의 등장

  • YARN, Spark

• SQL의 컴백: Hive, Presto등이 등장

  • Hive
    • MapReduce위에서 구현됨. Throughput에 초점. 대용량 ETL에 적합
  • Presto
    • Low latency에서 초점. 메모리를 주로 사용. Adhoc 쿼리에 적합
    • AWS Athena가 Presto 기반

Spark 소개

• 버클리 대학의 AMPLab에서 아파치 오픈소스 프로젝트로 2013년 시작
  • 나중에 Databricks라는 스타트업 창업
• 하둡의 뒤를 잇는 2세대 빅데이터 기술
  • YARN등을 분산환경으로 사용
  • Scala로 작성됨
• 빅데이터 처리 관련 다양한 기능 제공

Spark 3.0의 구성

• Spark Core
• Spark SQL
• Spark ML
  • Spark MLlib
• Spark Streaming
• Spark GraphX

Spark vs. MapReduce

• Spark은 기본적으로 메모리 기반
  • 메모리가 부족해지면 디스크 사용
  • MapReduce는 디스크 기반
• MapReduce는 하둡(YARN)위에서만 동작
  • Spark은 하둡(YARN)이외에도 다른 분산 컴퓨팅 환경 지원 (K8s, Mesos)
• MapReduce는 키와 밸류 기반 데이터 구조만 지원
  • Spark은 판다스 데이터프레임과 개념적으로 동일한 데이터 구조 지원
• Spark은 다양한 방식의 컴퓨팅을 지원
  • 배치 데이터 처리, 스트림 데이터 처리, SQL, 머신 러닝, 그래프 분석

Spark 프로그래밍 API

• RDD (Resilient Distributed Dataset)
  • 로우레벨 프로그래밍 API로 세밀한 제어가 가능
  • 하지만 코딩 복잡도 증가
• DataFrame & Dataset (판다스의 데이터프레임과 흡사)
  • 하이레벨 프로그래밍 API로 점점 많이 사용되는 추세
  • 구조화 데이터 조작이라면 보통 Spark SQL을 사용
  • DataFrame/Dataset이 꼭 필요한 경우는?
    • ML 피쳐 엔지니어링을 하거나 Spark ML을 쓰는 경우
    • SQL만으로 할 수 없는 일의 경우

Spark SQL

• Spark SQL은 구조화된 데이터 처리를 SQL로 처리
• 데이터 프레임을 SQL로 처리 가능
  • 데이터프레임은 테이블처럼 sql로 처리 가능
  • 판다스도 동일 기능 제공
• Hive 쿼리 보다 최대 100배까지 빠른 성능을 보장
  • 사실은 그렇지 않음. Hive도 그 사이에 메모리를 쓰는 걸로 발전
    • Hive: 디스크 -> 메모리
    • Spark: 메모리 -> 디스크
    • Presto: 메모리 -> 디스크

Spark ML

• 머신러닝 관련 다양한 알고리즘, 유틸리티로 구성된 라이브러리
• Classification, Regression, Clustering, Collaborative Filtering, …
  • 전체 리스트는 링크 참고. 딥러닝 지원은 미약
• RDD 기반과 데이터프레임 기반의 두 버전이 존재
  • spark.mllib vs. spark.ml
    • spark.mllib가 RDD 기반이고 spark.ml은 데이터프레임 기반
    • spark.mllib는 RDD위에서 동작하는 이전 라이브러리로 더 이상 업데이트가 안됨
  • 항상 spark.ml을 사용할 것!
    • import pyspark.ml

Spark ML의 장점

• 원스톱 ML 프레임웍!
  • 데이터프레임과 SparkSQL등을 이용해 전처리
  • Spark ML을 이용해 모델 빌딩
  • ML Pipeline을 통해 모델 빌딩 자동화
  • MLflow로 모델 관리하고 서빙 (MLOps)
• 대용량 데이터도 처리 가능!

Spark 데이터 시스템 사용 예들

• 기본적으로 대용량 데이터 배치 처리, 스트림 처리, 모델 빌딩
  • 예 1) 대용량 비구조화된 데이터 처리하기 (ETL 혹은 ELT)
  • 예 2) ML 모델에 사용되는 대용량 피쳐 처리 (배치/스트림)
  • 예 3) Spark ML을 이용한 대용량 훈련 데이터 모델 학습
• 대용량 비구조화된 데이터 처리하기 (Hive의 대체 기술)
  • ETL 혹은 ELT
• ML 모델에 사용되는 대용량 피쳐 처리

Spark 프로그램 실행 환경

• 개발/테스트/학습 환경 (Interactive Clients)

  • 노트북 (주피터, 제플린)
  • Spark Shell

• 프로덕션 환경 (Submit Job)

  • spark-submit (command-line utility): 가장 많이 사용됨
  • 데이터브릭스 노트북:
    • 노트북 코드를 주기적으로 실행해주는 것이 가능
  • REST API:
    • Spark Standalone 모드에서만 가능
    • API를 통해 Spark 잡을 실행
    • 실행코드는 미리 HDFS등의 파일 시스템에 적재되어 있어야함

      Spark 프로그램의 구조

• Driver

  • 실행되는 코드의 마스터 역할 수행 (YARN의 Application Master)

• Executor

  • 실제 태스크를 실행해주는 역할 수행 (YARN의 컨테이너)

• Driver:

  • 사용자 코드를 실행하며 실행 모드(client, cluster)에 따라 실행되는 곳이 달라짐
  • 코드를 실행하는데 필요한 리소스를 지정함 --num-executors, --executor-cores, --executor-memory
  • SparkSession을 만들어 Spark 클러스터와 통신 수행
    • Cluster Manager (YARN의 경우 Resource Manager)
      • Executor (YARN의 경우 Container)
  • 사용자 코드를 실제 Spark 태스크로 변환해 Spark 클러스터에서 실행

• Executor:

  • 실제 태스크를 실행해주는 역할 수행 (JVM): Transformations, Actions
  • YARN에서는 Container가 됨

Spark 클러스터 매니저 옵션

• local[n]:
  • 개발/테스트용
    • Spark Shell, IDE, 노트북
  • 하나의 JVM이 클러스터로 동작
    • Driver와 하나의 Executor 실행
  • n은 코어의 수
    • Executor의 스레드 수가 됨
  • local[*]는 무엇일까?
    • 컴퓨터에 있는 모든 코어 사용
• YARN
  • 두 개의 실행 모드가 존재: Client vs. Cluster
  • Client 모드: Driver가 Spark 클러스터 밖에서 동작
    • YARN 기반 Spark 클러스터를 바탕으로 개발/테스트 등을 할 때 사용
  • Cluster 모드: Driver가 Spark 클러스터 안에서 동작
    • 하나의 Container 슬롯을 차지
    • 실제 프로덕션 운영에 사용되는 모드

Spark 클러스터 매니저와 실행 모델 요약

• 데이터 자산 메타 정보 중앙 저장소
• 데이터 거버넌스의 첫 걸음
  • 많은 회사에서 데이터 카탈로그를 데이터 거버넌스 툴로 사용하거나 데이터 카탈로그 위에 커스텀 기능을 구현
• 데이터 카탈로그의 중요한 기능
  • (반)자동화된 메타 데이터 수집!
  • 데이터 보안! 보통 메타 데이터만 읽어옴

데이터 자산의 종류

• 테이블 (데이터베이스)
• 대시보드
• 문서/메세지 (슬랙, JIRA, Github, …)
• ML 피쳐
• 데이터 파이프라인
• 사용자 (HR 시스템)

데이터 카탈로그 : 데이터 자산의 효율적인 관리 프레임워크

• 다양한 관점에서 데이터를 조직적으로 관리
• 비지니스/데이터 용어 vs. 태그
• 데이터 오너 (Business & Technical)
• 표준화된 문서 템플릿

일반적인 데이터 카탈로그 아키텍처

데이터 카탈로그 주요 기능

• 주요 데이터 플랫폼 지원
• 비지니스 용어집 (Business Glossary)
• 주석/문서/태그 등 협업 기능
• 데이터 리니지
• 데이터 모니터링, 감사, 트레이싱
• 강력한 검색 기능 (통합 검색, NLP 검색)
• 데이터 추천 기능
• 데이터 유저 퍼소나 (예: 마케팅 분석가)

주요 데이터 플랫폼 지원

• Data Warehouses & Data Lakes: Redshift, Snowflake, BigQuery
• BI Tools: Looker, Tableau, Redash, Power BI, Mode, Superset
• ELT: DBT, Spark, Hive, PrestoDB
• ETL Orchestration: Airflow
• NoSQL and others
  • Cassandra, Druid, Elastic Search, Kafka Schema Registry, CSV
• Users: Azure AD, LDAP, …

비지니스 용어집 (Business Glossary)

• 권한이 있는 사람만 용어 정의가 가능
• 계층구조로 관리할 수 있다면 더 유용
  • DataHub의 경우 terms와 terms group 존재

비지니스 용어와 Entity 연결

• 나중에 다른 entity등과 연결 가능

협업 - 문서화 표준 제공

데이터 리니지

• Dataset-to-dataset
  • 보통 SQL 파싱으로 일어남
• Pipeline
  • 입력 데이터셋 -> Data Pipeline -> 출력 데이터셋
  • Airflow에 lineage backend라는 것이 존재
• Dashboard-to-chart
  • 하나의 차트가 여러 대시보드에 소속가능하기에 필요한 리니지
• Chart-to-dataset
• Job-to-dataflow
  • DBT에 특별한 리니지

데이터 거버넌스 관점에서 데이터 카탈로그의 중요성

• 우리가 갖고 있는 데이터 자산에 대한 통합 뷰를 제공
• 생산성 증대: 설문이나 데이터 티켓의 감소로 확인
• 위험 감소: 잘못된 결정과 개인정보등의 전파 방지
• 인프라 비용 감소: 불필요한 정보의 생성 방지와 안 쓰이는 데이터셋 삭제
• 데이터 티켓 감소
• 데이터 변경으로 인한 이슈 감소
  • 컬럼 레벨 리니지와 CI/CD 프로세스 연동

데이터 카탈로그 제품 서베이

데이터 카탈로그 트렌드

데이터 카탈로그 툴

• 상용제품
  • Alation, Collibra
  • Atlan, Select Star, Great Expectations
• 오픈소스
  • Amundsen (Lyft), DataHub (LinkedIn)
  • AcrylData (DataHub를 상용화)
• 클라우드 서비스
  • AWS Glue Data Catalog
  • Google Cloud Data Catalog
  • Microsoft Azure Data Catalog (Purview Data Catalog로 통합 중)
• 자체 툴
  • DataBook (Uber)
  • DataPortal (Airbnb)

ETL을 하는 이유는 결국 ELT를 하기 위함이며 이 때 데이터 품질 검증이 중요해짐

데이터 품질의 중요성 증대

• 입출력 체크
• 더 다양한 품질 검사
• 리니지 체크
• 데이터 히스토리 파악
• 데이터 품질 유지 -> 비용/노력 감소와 생산성 증대의 지름길

Database Normalization (정규화)

• 데이터베이스를 좀더 조직적이고 일관된 방법으로 디자인하려는 방법
• 데이터베이스 정합성을 쉽게 유지하고 레코드들을 수정/적재/삭제를 용이하게 하는 것
• Normalization에 사용되는 개념
  • Primary Key
  • Composite Key
  • Foreign Key

    1NF (First Normal Form)

• 한 셀에는 하나의 값만 있어야함
• Primary Key가 있어야함중복된 키나 레코드들이 없어야함

-> 목표는 중복을 제거하고 atomicity를 갖는 것

2NF (First Normal Form)

• 일단 1NF를 만족해야함
• 다음으로Primary Key를 중심으로 의존결과를 알 수 있어야함
• 부분적인 의존도가 없어야함
  • 즉 모든 부가 속성들은 Primary key를 가지고 찾을 수 있어야함
  • That is, all non-key attributes are fully dependent on a primary key

3NF (Third Normal Form)

• 일단 2NF를 만족해야함
• 전이적 부분 종속성을 없어야함
• 2NF의 예에서 state_code과 home_state가 같이 Employees 테이블에 존재

Slowly Changing Dimensions (SCD)

• DW나 DL에서는 모든 테이블들의 히스토리를 유지하는 것이 중요함
  • 보통 두 개의 timestamp 필드를 갖는 것이 좋음
    • created_at (생성시간으로 한번 만들어지면 고정됨)
    • updated_at (꼭 필요 마지막 수정 시간을 나타냄)
• 이 경우 컬럼의 성격에 따라 어떻게 유지할지 방법이 달라짐

SCD Type 0

• 한번 쓰고 나면 바꿀 이유가 없는 경우들
• 한번 정해지면 갱신되지 않고 고정되는 필드들
• 예) 고객 테이블이라면 회원 등록일, 제품 첫 구매일

  SCD Type 1

• 데이터가 새로 생기면 덮어쓰면 되는 컬럼들
• 처음 레코드 생성시에는 존재하지 않았지만 나중에 생기면서 채우는 경우
• 예) 고객 테이블이라면 연간소득 필드

  SCD Type 2

• 특정 entity에 대한 데이터가 새로운 레코드로 추가되어야 하는 경우
• 예) 고객 테이블에서 고객의 등급 변화
  • tier라는 컬럼의 값이 “regular”에서 “vip”로 변화하는 경우
  • 변경시간도 같이 추가되어야함

SCD Type 3

• SCD Type 2의 대안으로 특정 entity 데이터가 새로운 컬럼으로 추가되는 경우
• 예) 고객 테이블에서 tier라는 컬럼의 값이 “regular”에서 “vip”로 변화하는 경우
  • previous_tier라는 컬럼 생성
  • 변경시간도 별도 컬럼으로 존재해야함

SCD Type 4

• 특정 entity에 대한 데이터를 새로운 Dimension 테이블에 저장하는 경우
• SCD Type 2의 변종
• 예) 별도의 테이블로 저장하고 이 경우 아예 일반화할 수도 있음

dbt란 무엇인가?

• Data Build Tool (https://www.getdbt.com/)
  • ELT용 오픈소스: In-warehouse data transformation (이미 웨어하우스에 들어옴)
  • dbt Labs라는 회사가 상용화 ($4.2B valuation)
  • Analytics Engineer라는 말을 만들어냄
• 다양한 데이터 웨어하우스를 지원
  • Redshift, Snowflake, Bigquery, Spark
• 클라우드 버전도 존재
  • dbt Cloud

dbt 구성 컴포넌트

• 데이터 모델 (models)
  • 테이블들을 몇개의 티어로 관리
    • 일종의 CTAS (SELECT 문들), Lineage 트래킹
• Table, View, CTE 등등
  • 데이터 품질 검증 (tests)
  • 스냅샷 (snapshots)

다음과 같은 요구조건을 달성해야한다면?

• 데이터 변경 사항을 이해하기 쉽고 필요하다면 롤백 가능
• 데이터간 리니지 확인 가능
• 데이터 품질 테스트 및 에러 보고
• Fact 테이블의 증분 로드 (Incremental Update)
• Dimension 테이블 변경 추적 (히스토리 테이블)
• 용이한 문서 작성

무슨 ELT 작업을 해볼까요?

• Redshift 사용
• AB 테스트 분석을 쉽게 하기 위한 ELT 테이블을 만들어보자
• 입력 테이블:
  • user_event, user_variant, user_metadata
• 생성 테이블: Variant별 사용자별 일별 요약 테이블
  • variant_id, user_id, datestamp, age, gender,
  • 총 impression, 총 click, 총 purchase, 총 revenue

입력 데이터들

• Production DB에 저장되는 정보들을 Data Warehouse로 적재했다고 가정
• raw_data.user_event
  • 사용자/날짜/아이템별로 impression이 있는 경우 그 정보를 기록하고 impression으로부터 클릭, 구매, 구매시 금액을 기록. 실제 환경에서는 이런 aggregate 정보를 로그 파일등의 소스(하나 이상의 소스가 될 수도 있음)로부터 만들어내는 프로세스가 필요함
• raw_data.user_variant
  • 사용자가 소속한 AB test variant를 기록한 파일 (control vs. test)
• raw_data.user_metadata
  • 사용자에 관한 메타 정보가 기록된 파일 (성별, 나이 등등)

입력데이터: raw_data.user_event

 CREATE TABLE raw_data.user_event (
    user_id int,
    datestamp timestamp,
    item_id int,
    clicked int,
    purchased int,
    paidamount int
 );

입력데이터: raw_data.user_variant

 CREATE TABLE raw_data.user_variant (
    user_id int,
    variant_id varchar(32)   -- control vs. test
 );

• 보통은 experiment와 variant 테이블이 별도로 존재함
• 그리고 위의 테이블에도 언제 variant_id로 소속되었는지 타임스탬프 필드가 존재하는 것이 일반적

입력데이터: raw_data.user_metadata

 CREATE TABLE raw_data.user_metadata (
    user_id int,
    age varchar(16),
    gender varchar(16)
 );

• 사용자별 메타정보: 이를 이용해 다양한 각도에서 AB 테스트 결과를 분석해볼 수 있음

Fact 테이블과 Dimension 테이블

• Fact 테이블: 분석의 초점이 되는 양적 정보를 포함하는 중앙 테이블

  • 일반적으로 매출 수익, 판매량, 이익과 같은 측정 항목 포함. 비즈니스 결정에 사용
  • Fact 테이블은 일반적으로 외래 키를 통해 여러 Dimension 테이블과 연결됨
  • 보통 Fact 테이블의 크기가 훨씬 더 큼

• Dimension 테이블: Fact 테이블에 대한 상세 정보를 제공하는 테이블

  • 고객, 제품과 같은 테이블로 Fact 테이블에 대한 상세 정보 제공
  • Fact 테이블의 데이터에 맥락을 제공하여 다양한 방식으로 분석 가능하게 해
  • Dimension 테이블은 primary key를 가지며, fact 테이블에서 참조 (foreign key)
  • 보통 Dimension 테이블의 크기는 훨씬 더 작음

입력 데이터 요약

• user_event, user_variant, user_metadata

최종 생성 데이터 (ELT 테이블)

• SELECT로 표현하면 아래와 같음

  SELECT 
    variant_id,
    ue.user_id,
    datestamp,
    age,
    gender,
    COUNT(DISTINCT item_id) num_of_items, -- 총 impression
    COUNT(DISTINCT CASE WHEN clicked THEN item_id END) num_of_clicks, -- 총 click
    SUM(purchased) num_of_purchases,  -- 총 purchase
    SUM(paidamount) revenue                   -- 총 revenue
  FROM raw_data.user_event ue 
  JOIN raw_data.user_variant uv ON ue.user_id = uv.user_id
  JOIN raw_data.user_metadata um ON uv.user_id = um.user_id
  GROUP by 1, 2, 3, 4, 5;

DBT 설치와 환경 설정

• dbt 사용절차
• dbt 설치
  • dbt Cloud vs. dbt Core
  • git을 보통 사용함
• dbt 환경설정
• Connector 설정
  • Connector가 바로 바탕이 되는 데이터 시스템 (Redshift, Spark, …)
• 데이터 모델링 (tier)
  • Raw Data -> Staging -> Core
• 테스트 코드 작성
• (필요하다면) Snapshot 설정

dbt Models: Input

Model이란?

• ELT 테이블을 만듬에 있어 기본이 되는 빌딩블록
  • 테이블이나 뷰나 CTE의 형태로 존재
• 입력,중간,최종 테이블을 정의하는 곳
  • 티어 (raw, staging, core, …)
  • raw => staging (src) => core

잠깐: View란 무엇인가?

• SELECT 결과를 기반으로 만들어진 가상 테이블
  • 기존 테이블의 일부 혹은 여러 테이블들을 조인한 결과를 제공함
  • CREATE VIEW 이름 AS SELECT …
• View의 장점
  • 데이터의 추상화: 사용자는 View를 통해 필요 데이터에 직접 접근. 원본 데이터를 알 필요가 없음
  • 데이터 보안: View를 통해 사용자에게 필요한 데이터만 제공. 원본 데이터 접근 불필요
  • 복잡한 쿼리의 간소화: SQL(View)를 사용하면 복잡한 쿼리를 단순화.
• View의 단점
  • 매번 쿼리가 실행되므로 시간이 걸릴 수 있음
  • 원본 데이터의 변경을 모르면 실행이 실패함

잠깐: CTE (Common Table Expression)

 WITH src_user_event AS (
    SELECT * FROM raw_data.user_event
 )
 SELECT
    user_id,
    datestamp,
    item_id,
    clicked,
    purchased,
    paidamount
 FROM
    src_user_event

Model 구성 요소

• Input
  • 입력(raw)과 중간(staging, src) 데이터 정의
  • raw는 CTE로 정의
  • staging은 View로 정의
• Output
  • 최종(core) 데이터 정의
  • core는 Table로 정의
• 이 모두는 models 폴더 밑에 sql 파일로 존재
  • 기본적으로는 SELECT + Jinja 템플릿과 매크로
  • 다른 테이블들을 사용 가능 (reference)
    • 이를 통해 리니지 파악

데이터 빌딩 프로세스

Model 빌딩 확인

• 해당 스키마 밑에 테이블 생성 여부 확인
• dbt run은 프로젝트 구성 다양한 SQL 실행
  • 이 SQL들은 DAG로 구성됨
• dbt run은 보통 다른 더 큰 명령의 일부로 실행
  • dbt test
  • dbt docs generate

dbt Models: Output

Materialization이란?

• 입력 데이터(테이블)들을 연결해서 새로운 데이터(테이블) 생성하는 것
  • 보통 여기서 추가 transformation이나 데이터 클린업 수행
• 4가지의 내장 materialization이 제공됨
• 파일이나 프로젝트 레벨에서 가능
• 역시 dbt run을 기타 파라미터를 가지고 실행

4가지의 Materialization 종류

• View
  • 데이터를 자주 사용하지 않는 경우
• Table
  • 데이터를 반복해서 자주 사용하는 경우
• Incremental (Table Appends)
  • Fact 테이블
  • 과거 레코드를 수정할 필요가 없는 경우
• Ephemeral (CTE)
  • 한 SELECT에서 자주 사용되는 데이터를 모듈화하는데 사용

잠깐 Jinja 템플릿이란?

• 파이썬이 제공해주는 템플릿 엔진으로 Flask에서 많이 사용
  • Airflow에서도 사용함
• 입력 파라미터 기준으로 HTML 페이지(마크업)를 동적으로 생성
• 조건문, 루프, 필터등을 제공

dbt compile vs. dbt run

• dbt compile은 SQL 코드까지만 생성하고 실행하지는 않음
• dbt run은 생성된 코드를 실제 실행함

Model 빌딩 확인

• 해당 스키마 밑에 테이블 생성 여부 확인
• Core 테이블들은 Table
• Staging 테이블들은 View

데이터 빌딩 프로세스

Seeds 소개

• 많은 dimension 테이블들은 크기가 작고 많이 변하지 않음
• Seeds는 이를 파일 형태로 데이터웨어하우스로 로드하는 방법
  • Seeds는 작은 파일 데이터를 지칭 (보통 csv 파일)
• dbt seed를 실행해서 빌드

Sources 소개

• 기본적으로 처음 입력이 되는 ETL 테이블을 대상으로 함
  • 별칭 제공
  • 최신 레코드 체크 기능 제공
• 테이블 이름들에 별명(alias)을 주는 것
  • 이를 통해 ETL단의 소스 테이블이 바뀌어도 뒤에 영향을 주지 않음
  • 추상화를 통한 변경처리를 용이하게 하는 것
  • 이 별명은 source 이름과 새 테이블 이름의 두 가지로 구성됨
    • 예) raw_data.user_metadata -> keeyong, metadata
• Source 테이블들에 새 레코드가 있는지 체크해주는 기능도 제공

Sources 최신성 (Freshness)

• 특정 데이터가 소스와 비교해서 얼마나 최신성이 떨어지는지 체크하는 기능
• dbt source freshness 명령으로 수행
• 이를 하려면 models/sources.yml의 해당 테이블 밑에 아래 추가

  sources:
  - name: keeyong
    schema: raw_data
    tables:
      - name: event
        identifier: user_event
        loaded_at_field: datestamp
        freshness:
          warn_after: { count: 1, period: hour }
          error_after: { count: 24, period: hour }

  DBT Snapshots

• Dimension 테이블은 성격에 따라 변경이 자주 생길 수 있음
• dbt에서는 테이블의 변화를 계속적으로 기록함으로써 과거 어느 시점이건 다시 돌아가서 테이블의 내용을 볼 수 있는 기능을 이야기함
  • 이를 통해 테이블에 문제가 있을 경우 과거 데이터로 롤백 가능
  • 다양한 데이터 관련 문제 디버깅도 쉬워짐

dbt의 스냅샷 처리 방법

• 먼저 snapshots 폴더에 환경설정이 됨
• snapshots을 하려면 데이터 소스가 일정 조건을 만족해야함
  • Primary key가 존재해야함
  • 레코드의 변경시간을 나타내는 타임스탬프 필요 (updated_at, modified_at 등등)
• 변경 감지 기준
  • Primary key 기준으로 변경시간이 현재 DW에 있는 시간보다 미래인 경우
• Snapshots 테이블에는 총 4개의 타임스탬프가 존재
  • dbt_scd_id, dbt_updated_at
  • valid_from, valid_to

DBT Tests 소개

• 데이터 품질을 테스트하는 방법
• 두 가지가 존재
  • 내장 일반 테스트 (“Generic”)
    • unique, not_null, accepted_values, relationships 등의 테스트 지원
    • models 폴더
  • 커스텀 테스트 (“Singular”)
    • 기본적으로 SELECT로 간단하며 결과가 리턴되면 “실패”로 간주
    • tests 폴더

Generic Tests 구현

• models/schema.yml 파일 생성

  version: 2
  models:
  - name: dim_user_metadata
    columns:
    - name: user_id
      tests:
      - unique
      - not_null

  Singular Tests 구현

• tests/dim_user_metadata.sql 파일 생성
  • Primary Key Uniqueness 테스트

    SELECT
    *
    FROM (
    SELECT
    user_id, COUNT(1) cnt
    FROM 
    {{ ref("dim_user_metadata") }}
    GROUP BY 1
    ORDER BY 2 DESC
    LIMIT 1
    )
    WHERE cnt > 1

DBT Documentation 소개

• 기본 철학은 문서와 소스 코드를 최대한 가깝게 배치하자는 것
• 문서화 자체는 두 가지 방법이 존재
  • 기존 .yml 파일에 문서화 추가 (선호되는 방식)
  • 독립적인 markdown 파일 생성
• 이를 경량 웹서버로 서빙
  • overview.md가 기본 홈페이지가 됨
  • 이미지등의 asset 추가도 가능

models 문서화 하기

• description 키를 추가: models/schema.yml, models/sources.yml

  version: 2
  models:
  - name: dim_user_metadata
  description: A dimension table with user metadata
    columns:
    - name: user_id
  description: The primary key of the table
      tests:
      - unique
      - not_null

• dbt docs generate
  • 사용자 권한이 더 있다면 Redshift로부터 더 많은 정보를 가져다가 보여줌
  • 결과 파일은 target/catalog.json 파일이 됨

DBT Expectations 소개

• Great Expectations에서 영감을 받아 dbt용으로 만든 dbt 확장판
  • https://github.com/calogica/dbt-expectations
• 설치 후 packages.yml에 등록

  packages:
  - package: calogica/dbt_expectations
   version: [">=0.7.0", "<0.8.0"]

• 보통은 앞서 dbt 제공 테스트들과 같이 사용
  • models/schema.yml

Airflow에서 발생되는 로그파일의 크기는 작지 않다. 이를 주기적으로 삭제하거나 백업하는 것에 대해 알아보자

Airflow 로그 위치

두 군데에 별도의 로그가 기록됨. 이를 주기적으로 삭제하거나 백업 (s3) 필요

[logging]
# The folder where airflow should store its log files
# This path must be absolute
base_log_folder = /var/lib/airflow/logs

[scheduler]
child_process_log_directory = /var/lib/airflow/logs/scheduler

Airflow 로그 위치 - docker compose

• docker compose로 실행된 경우 logs 폴더가 host volume의 형태로 유지

  volumes:
    - ${AIRFLOW_PROJ_DIR:-.}/dags:/opt/airflow/dags
    - ${AIRFLOW_PROJ_DIR:-.}/logs:/opt/airflow/logs

Dag를 실행하는 방법

• 주기적 실행: schedule로 지정
• 다른 Dag에 의해 트리거
  • Explicit Trigger: Dag A가 분명하게 Dag B를 트리거 (TriggerDagRunOperator)
  • Reactive Trigger: Dag B가 Dag A가 끝나기를 대기 (ExternalTaskSensor)
• 알아두면 좋은 상황에 따라 다른 태스크 실행 방식들
  • 조건에 따라 다른 태스크로 분기 (BranchPythonOperator)
  • 과거 데이터 Backfill시에는 불필요한 태스크 처리 (LatestOnlyOperator)
  • 앞단 태스크들의 실행상황
    • 어떤 경우에는 앞단이 실패해도 동작해야하는 경우가 있을 수 있음

두 가지 방법이 존재

• Explicit trigger
  • TriggerDagRunOperator
  • DAG A가 명시적으로 DAG B를 트리거
• Reactive trigger
  • ExternalTaskSensor
  • DAG B가 DAG A의 태스크가 끝나기를 대기
    • 이 경우 DAG A는 이 사실을 모름

TriggerDagRunOperator

• DAG A의 태스크를 TriggerDagRunOperator로 구현

  from airflow.operators.trigger_dagrun import TriggerDagRunOperator
  

trigger_B = TriggerDagRunOperator( task_id="trigger_B", trigger_dag_id="트리거하려는DAG이름" )

### 참고: Jinja Template이란?
- Jinja 템플릿은 Python에서 널리 사용되는 템플릿 엔진
   - Django 템플릿 엔진에서 영감을 받아 개발
   - Jinja를 사용하면 프레젠테이션 로직과 애플리케이션 로직을 분리하여 동적으로 HTML 생성
   - Flask에서 사용됨
- 변수는 이중 중괄호 {{ }}로 감싸서 사용

안녕하세요, {{ name }}님!

{% for item in items %}
• {{ item }}
{% endfor %}

# BashOperator를 사용하여 템플릿 작업 정의
task1 = BashOperator(
    task_id='task1',
    bash_command='echo "{{ ds }}"',
    dag=dag
    )

• 예 2) 파라미터 등으로 넘어온 변수를 쉽게 사용 가능

  # 동적 매개변수가 있는 다른 템플릿 작업 정의
  task2 = BashOperator(
    task_id='task2',
     bash_command='echo "안녕하세요, {{ params.name }}!"',
    params={'name': 'John'},  # 사용자 정의 가능한 매개변수
    dag=dag
    ) 

  참고: Airflow에서 사용 가능한 Jinja 변수들 몇개 살펴보기

• {{ ds }}
• {{ ds_nodash }}
• {{ ts }}
• {{ dag }}
• {{ task }}
• {{ dag_run }}
• {{ var.value }}: {{ var.value.get('my.var', 'fallback') }}
• {{ var.json }}: {{ var.json.my_dict_var.key1 }}
• {{ conn }}: {{ conn.my_conn_id.login }}, {{ conn.my_conn_id.password }}

# TriggerDagRunOperator 예시

from airflow.operators.trigger_dagrun import TriggerDagRunOperator
trigger_B = TriggerDagRunOperator(
    task_id="trigger_B",
    trigger_dag_id="트리거하려는DAG이름",
    conf={ 'path': '/opt/ml/conf' }, 
# DAG B에 넘기고 싶은 정보. DAG B에서는 Jinja 템플릿(dag_run.conf["path"])으로 접근 가능.
# DAG B PythonOperator(**context)에서라면 kwargs['dag_run'].conf.get('path')

    execution_date="{{ ds }}",  # Jinja 템플릿을 통해 DAG A의 execution_date을 패스
    reset_dag_run=True, # True일 경우 해당 날짜가 이미 실행되었더라는 다시 재실행
    wait_for_completion=True    # DAG B가 끝날 때까지 기다릴지 여부를 결정. 디폴트값은 False
    )

airflow.cfg의 dag_run_conf_overrides_params가 True로 설정되어 있어야함

Sensor란 무엇인가?

• Sensor는 특정 조건이 충족될 때까지 대기하는 Operator
• Sensor는 외부 리소스의 가용성이나 특정 조건의 완료와 같은 상황 동기화에 유용
• Airflow는 몇 가지 내장 Sensor를 제공
  • FileSensor: 지정된 위치에 파일이 생길 때까지 대기
  • HttpSensor: HTTP 요청을 수행하고 지정된 응답이 대기
  • SqlSensor: SQL 데이터베이스에서 특정 조건을 충족할 때까지 대기
  • TimeSensor: 특정 시간에 도달할 때까지 워크플로우를 일시 중지
  • ExternalTaskSensor: 다른 Airflow DAG의 특정 작업 완료를 대기
• 기본적으로 주기적으로 poke를 하는 것
  • worker를 하나 붙잡고 poke간에 sleep를 할지 아니면 worker를 릴리스하고 다시 잡아서 poke를 할지 결정해주는 파라미터가 존재: mode
    • mode의 값은 reschedule 혹은 poke가 됨

_-> reschedule이 worker utilization적으로 유용, poke는 worker 하나가 그냥 낭비될 수 있음 _

ExternalTaskSensor

-> 왠만하면 쓰지 않는게 좋음

• DAG B의 ExternalTaskSensor 태스크가 DAG A의 특정 태스크가 끝났는지 체크함
  • 먼저 동일한 schedule_interval을 사용
  • 이 경우 두 태스크들의 Execution Date이 동일해야함. 아니면 매칭이 안됨!

from airflow.sensors.external_task import ExternalTaskSensor

waiting_for_end_of_dag_a = ExternalTaskSensor(
    task_id='waiting_for_end_of_dag_a',
    external_dag_id='DAG이름',
    external_task_id='end',
    timeout=5*60,
    mode='reschedule'    
    )

• 만일 DAG A와 DAG B가 서로 다른 schedule interval을 갖는다면 ?
  • 예를 들어 DAG A가 DAG B보다 5분 먼저 실행된다면?
    • execution_delta를 사용
    • execution_date_fn을 사용하면 조금더 복잡하게 컨트롤 가능
  • 만일 두개의 DAG가 서로 다른 frequency를 갖고 있다면 이 경우 ExternalTaskSensor는 사용불가

    from airflow.sensors.external_task import ExternalTaskSensor
    waiting_for_end_of_dag_a = ExternalTaskSensor(
    task_id='waiting_for_end_of_dag_a',
    external_dag_id='DAG이름',
    external_task_id='end',
    timeout=5*60,
    mode='reschedule',
    execution_delta=timedelta(minutes=5)
    )    

    BranchPythonOperator

• 상황에 따라 뒤에 실행되어야할 태스크를 동적으로 결정해주는 오퍼레이터
  • 미리 정해준 Operator들 중에 선택하는 형태로 돌아감
• TriggerDagOperator 앞에 이 오퍼레이터를 사용하는 경우도 있음

  from airflow.operators.python import BranchPythonOperator
  # 상황에 따라 뒤에 실행되어야 하는 태스크를 리턴
  def skip_or_cont_trigger():
    if Variable.get("mode", "dev") == "dev":
    return []
    else:
    return ["trigger_b"]
  # "mode"라는 Variable의 값이 "dev"이면 trigger_b 태스크를 스킵
  branching = BranchPythonOperator(
    task_id='branching',
    python_callable=skip_or_cont_trigger,
    )

LatestOnlyOperator

• Time-sensitive한 태스크들이 과거 데이터의 backfill시 실행되는 것을 막기 위함
• 현재 시간이 지금 태스크가 처리하는 execution_date보다 미래이고 다음 execution_date보다는 과거인 경우에만 뒤로 실행을 이어가고 아니면 여기서 중단됨
  • t1 >> t3 >> [t2, t4]

from airflow.operators.latest_only import LatestOnlyOperator
from airflow.operators.empty import EmptyOperator
with DAG(
    dag_id='latest_only_example',
    schedule=timedelta(hours=48),    # 매 48시간마다 실행되는 DAG로 설정
    start_date=datetime(2023, 6, 14),
    catchup=True) as dag:

    t1 = EmptyOperator(task_id='task1')
     t2 = LatestOnlyOperator(task_id = 'latest_only')
     t3 = EmptyOperator(task_id='task3')
    t4 = EmptyOperator(task_id='task4')
    t1 >> t2 >> [t3, t4]

2023-06-18일 오후 늦게 실행 과거 날짜가 catchup=True 때문에 실행되었지만 스킵됨

Trigger Rules이란?

• Upstream 태스크의 성공실패 상황에 따라 뒷단 태스크의 실행여부를 결정하고 싶다면?
  • 보통 앞단이 하나라도 실패하면 뒷 단의 태스크는 실행불가
• Operator에 trigger_rule이란 파라미터로 결정 가능
  • trigger_rule은 태스크에 주어지는 파라미터로 다음과 같은 값이 가능
  • all_success (기본값), all_failed, all_done, one_failed, one_success, none_failed, none_failed_min_one_success

Trigger Rule의 가능값 (airflow.utils.trigger_rule.TriggerRule)

• ALL_SUCCESS: (default) all parents have succeeded
• ALL_FAILED: all parents are in a failed or upstream_failed state
• ALL_DONE: all parents are done with their execution (성공실패 여부와 관계없이)
• ONE_FAILED:
  • fires as soon as at least one parent has failed, it does not wait for all parents to be done
• ONE_SUCCESS:
  • fires as soon as at least one parent succeeds, it does not wait for all parents to be done
• NONE_FAILED:
  • all parents have not failed (or upstream_failed) i.e. all parents have succeeded or been skipped
• NONE_FAILED_MIN_ONE_SUCCESS
  • one parent at least is done but none failed

Trigger Rule 사용 예

from airflow.utils.trigger_rule import TriggerRule

with DAG("trigger_rules", default_args=default_args, schedule=timedelta(1)) as dag:
    t1 = BashOperator(task_id="print_date", bash_command="date")
    t2 = BashOperator(task_id="sleep", bash_command="sleep 5")
    t3 = BashOperator(task_id="exit", bash_command="exit 1")
    t4 = BashOperator(
    task_id='final_task',
    bash_command='echo DONE!',
    trigger_rule=TriggerRule.ALL_DONE
    )
    [t1, t2, t3] >> t4

팁: Airflow 메타데이터 DB 내용 살펴보기

• airflow:airflow로 Postgres에 로그인 가능

  docker exec -it learn-airflow-airflow-webserver-1 sh

• 그 다음에 아래 명령 수행

  psql -h postgres

• psql shell에서 아래 명령 수행

  \dt
  select * FROM dag_run LIMIT 10;
  DELETE FROM dag_run WHERE dag_id = '기록을삭제하고싶은DAG’;

  태스크 그룹핑의 필요성

• 태스크 수가 많은 DAG라면 태스크들을 성격에 따라 관리하고 싶은 니즈 존재
  • SubDAG이 사용되다가 Airflow 2.0에서 나온 Task Grouping으로 넘어가는 추세
    • SubDAG를 비슷한 일을 하는 태스크들을 SubDAG라는 Child Dag로 만들어서 관리
• 다수의 파일 처리를 하는 DAG라면
  • 파일 다운로드 태스크들과 파일 체크 태스크와 데이터 처리 태스크들로 구성

예제 살펴보기 - 소스코드

• Learn Task Groups
  • TaskGroup 안에 TaskGroup nesting 가능
  • TaskGroup도 태스크처럼 실행 순서 정의 가능

    from airflow.utils.task_group import TaskGroup
    start = EmptyOperator(task_id="start")
    with TaskGroup("Download", tooltip="Tasks for downloading data") as section_1:
    task_1 = EmptyOperator(task_id="task_1")
    task_2 = BashOperator(task_id="task_2", bash_command='echo 1')
    task_3 = EmptyOperator(task_id="task_3")
    task_1 >> [task_2, task_3]
    start >> section_1

    

Dynamic Dag란 무엇인가?

• 템플릿과 YAML을 기반으로 DAG를 동적으로 만들어보자
  • Jinja를 기반으로 DAG 자체의 템플릿을 디자인하고 YAML을 통해 앞서 만든 템플릿에 파라미터를 제공
• 이를 통해 비슷한 DAG를 계속해서 매뉴얼하게 개발하는 것을 방지
• DAG를 계속해서 만드는 것과 한 DAG안에서 태스크를 늘리는 것 사이의 밸런스 필요
  • 오너가 다르거나 태스크의 수가 너무 커지는 경우 DAG를 복제해나가는 것이 더 좋음

Dynamic Dag의 기본적인 아이디어

Docker 기반 Airflow 설정

docker-compose.yaml 수정

• _PIP_ADDITIONAL_REQUIREMENTS 수정
• data 폴더를 호스트 폴더에서 만들고 볼륨으로 공유: 임시 데이터를 저장할 폴더 이를 docker volume으로 지정해서 나중에 디버깅에 사용

  environment:
    AIRFLOW_VAR_DATA_DIR: /opt/airflow/data
    _PIP_ADDITIONAL_REQUIREMENTS: ${_PIP_ADDITIONAL_REQUIREMENTS:- yfinance pandas numpy 
  oauth2client gspread}
  volumes:
    …
    - ${AIRFLOW_PROJ_DIR:-.}/data:/opt/airflow/data
  airflow-init:
       …
       mkdir -p /sources/logs /sources/dags /sources/plugins /sources/data
       chown -R "${AIRFLOW_UID}:0" /sources/{logs,dags,plugins,data

• :-는 if문과 비슷한 역할을 함

• 판다스, 구글시트에 연동할 때 필요한 것들

• 새로운 pip 추가, 볼륨 추가, airflow-init 추가 권한 추가

  다음 명령을 수행. Detached 모드로 실행하려면 -d 옵션 지정 (-f 옵션도 존재)

• docker compose up -d

  http://localhost:8080으로 웹 UI 로그인

• 아이디 비번 airflow:airflow 사용
• 앞서 설정한 DATA_DIR이란 변수는 Admin => Variables에 안 보임 DAG과 Airflow 환경 정보들은 Postgres의 Named Volume으로 유지되고 있음 환경변수로 설정한 것들은 Web UI에서는 안 보이지만 프로그램에서는 사용가능 $ docker exec -it learn-airflow-airflow-scheduler-1 airflow variables get DATA_DIR /opt/airflow/data

  Variables/Connections 설정을 어떻게 관리하는 것이 좋을까?

• 이를 docker-compose.yaml에서 환경변수로 설정하는 것이 좋음

고민 포인트: Airflow 실행환경 관리방안

기타 환경설정값들 (Variables, Connections 등등)을 어떻게 관리/배포할까?

• 보통 docker-compose.yml 파일에서 아래 부분에 정의 x-airflow-common: &airflow-common … environment: &airflow-common-env AIRFLOW_VAR_DATA_DIR: /opt/airflow/data AIRFLOW_CONN_TEST_ID: test_connection

-> 데브옵스 팀과 환경설정에 대해 문의해보는 것이 좋다 (소스코드는 엔지니어의 일, 데브옵스 팀이 없으면 깃헙 리포에 두고 알아서 하면 됨) -> 환경변수가 아니라 별도 Secrets credentials 전용 백엔드라는 것을 사용하기도 함 (이게 조금 더 안전한 방법)

어디까지 Airflow 이미지로 관리하고 무엇을 docker-compose.yml에서 관리할지 생각

• 이는 회사마다 조금씩 다름
• Airflow 자체 이미지를 만들고 거기에 넣을지? 이 경우 환경변수를 자체 이미지에 넣고 이를 docker-compose.yaml 파일에서 사용 x-airflow-common: &airflow-common image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.5.1}
• 아니면 docker-compose.yaml에서 환경변수를 직접 설정

-> AIRFLOW_IMAGE_NAME 환경변수가 정의되어 있다면 그걸 사용하고 아니면 기본값으로 apache/airflow:2.5.1

DAG 코드도 마찬가지

• Airflow image로 DAG 코드를 복사하여 만드는 것이 좀더 깔끔
• 아니면 docker-compose에서 host volume 형태로 설정 이는 개발/테스트용으로 좀더 적합

팁: .airflowignore

Airflow의 DAG 스캔 패턴은?

• dags_folder가 가리키는 폴더를 서브폴더들까지 다 스캔해서 DAG 모듈이 포함된 모든 파이썬 스크립트를 실행해서 새로운 DAG를 찾게 되며 이는 가끔 사고로 이어짐
• Airflow가 의도적으로 무시해야 하는 DAG_FOLDER의 디렉터리 또는 파일을 지정
• .airflowignore의 각 줄은 정규식 패턴으로 지정하며 매칭되는 파일들은 무시됨

  ex)
  project_a
  tenant_[\d]

• 위의 경우 아래 파일들이 무시됨 project_a_dag_1.py, TESTING_project_a.py, tenant_1.py, project_a/dag_1.py

Summary 테이블 구현

Summary table: 간단한 DAG 구현을 살펴보기

• Build_Summary.py: MAU(Monthly Active User) 요약 테이블을 만들어보자
• 이 부분을 dbt로 구현하는 회사들도 많음 (Analytics Engineer) https://www.getdbt.com/ 별도 강의에서 다룰 예정

Summary table: 이번에 사용자별 channel 정보를 요약해보자

• 앞서와 비슷하게 PythonOperator를 만들고 아래처럼 params 파라미터를 설정

    params = {
        'schema' : 'keeyong',
        'table': 'channel_summary',
        'sql' : """SELECT
      DISTINCT A.userid,
        FIRST_VALUE(A.channel) over(partition by A.userid order by B.ts rows between unbounded preceding and unbounded following) AS First_Channel,
        LAST_VALUE(A.channel) over(partition by A.userid order by B.ts rows between unbounded preceding and unbounded following) AS Last_Channel
        FROM raw_data.user_session_channel A
        LEFT JOIN raw_data.session_timestamp B ON A.sessionid = B.sessionid;"""
    }

  CTAS 부분을 아예 별도의 환경설정 파일로 떼어내면 어떨까?

• 환경 설정 중심의 접근 방식
  • config 폴더를 생성
  • 그 안에 써머리 테이블별로 하나의 환경설정 파일 생성
    • 파이썬 dictionary 형태로 유지할 것이라 .py 확장자를 가져야함
      • 이렇게 하면 비개발자들이 사용할 때 어려움을 덜 느끼게 됨
      • 그러면서 더 다양한 테스트를 추가

NPS 써머리 테이블을 만들어 보자

• NPS란? Net Promoter Score

  • 10점 만점으로 '주변에 추천하겠는가?'라는 질문을 기반으로 고객 만족도를 계산

  • 10, 9점 추천하겠다는 고객(promoter)의 비율에서 0-6점의 불평고객(detractor)의 비율을 뺀 것이 NPS

    • 7, 8점은 아예 계산에 안 들어감

• 각자스키마.nps 테이블 혹은 raw_data.nps 테이블 기준으로 일별 nps 써머리 생성
  • 먼저 SQL을 만들어보자

일별 NPS 계산 SQL

 SELECT LEFT(created_at, 10) AS date,
  ROUND(
    SUM(
      CASE
        WHEN score >= 9 THEN 1 
        WHEN score <= 6 THEN -1
      END
    )::float*100/COUNT(1), 2
  ) nps
 FROM keeyong.nps
 GROUP BY 1
 ORDER BY 1;

NPS Summary를 주기적으로 요약 테이블로 만들기

• CTAS 부분을 아예 별도의 파일로 떼어내면 어떨까?
  • 환경 설정 중심의 접근 방식
  • config/nps_summary.py

    {
    'table': 'nps_summary',
    'schema': 'keeyong',
    'main_sql': """SELECT …;""",
    'input_check': [ {
    'sql': 'SELECT COUNT(1) FROM keeyong.nps',
    'count': 150000
    } ],
    'output_check': [ {
    'sql': 'SELECT COUNT(1) FROM {schema}.temp_{table}',
    'count': 12
    } ],
    }

• 새로운 Operator와 helper 함수 구현
  • RedshiftSummaryOperator
  • build_summary_table
  • Build_Summary_v2.py
• 다른 방법은 dbt 사용하기
  • Analytics Engineering (ELT)

Slack 연동하기

• DAG 실행 중에 에러가 발생하면 그걸 지정된 슬랙 workspace의 채널로 보내기
• 이를 위해서 해당 슬랙 workspace에 App 설정이 필요
• 다음으로 연동을 위한 함수를 하나 만들고 (plugins/slack.py)
• 이를 태스크에 적용되는 default_args의 on_failure_callback에 지정

  from plugins import slack
    …
    default_args= {
  'on_failure_callback': slack.on_failure_callback,
  }

  먼저 어느 Workspace의 어느 Channel로 보낼 것인지 결정

• prgms-de라는 Workspace 밑에 DataAlert이라는 App 생성
• 이 App이 #data-alert이라는 채널에 메세지를 보낼 수 있게 설정

데이터 파이프라인 실패/경고를 슬랙으로 보내는 방법

• T016X1V5HBQ/B02QB4GGNQM/xone4l4N3gMLTQRnRBWYaZ9y를 “slack_url” Variable로 저장

• slack에 에러 메세지를 보내는 별도 모듈로 개발: slack.py

  • 이를 DAG 인스턴스를 만들 때 에러 콜백으로 지정

1. DAGs 폴더는 어디에 지정되는가?

a. 기본적으로는 Airflow가 설치된 디렉토리 밑의 dags 폴더가 되며 dags_folder 키에 저장됨

2. DAGs 폴더에 새로운 Dag를 만들면 언제 실제로 Airflow 시스템에서 이를 알게되나? 이 스캔 주기를 결정해주는 키의 이름이 무엇인가?

a. dag_dir_list_interval (기본값은 300 = 5분)

3. 이 파일에서 Airflow를 API 형태로 외부에서 조작하고 싶다면 어느 섹션을 변경해야하는가?

a. api 섹션의 auth_backend를 airflow.api.auth.backend.basic_auth로 변경

4. Variable에서 변수의 값이 encrypted가 되려면 변수의 이름에 어떤 단어들이 들어가야 하는데 이 단어들은 무엇일까? :)

a. password, secret, passwd, authorization, api_key, apikey, access_token

5. 이 환경 설정 파일이 수정되었다면 이를 실제로 반영하기 위해서 해야 하는 일은?

a. sudo systemctl restart airflow-webserver
b. sudo systemctl restart airflow-scheduler

6. Metadata DB의 내용을 암호화하는데 사용되는 키는 무엇인가?

a. fernet_key

Airflow와 타임존

airflow.cfg에는 두 종류의 타임존 관련 키가 존재

a. default_timezone b. default_ui_timezone

start_date, end_date, schedule

a. default_timezone에 지정된 타임존을 따름

execution_date와 로그 시간

a. 항상 UTC를 따름 b. 즉 execution_date를 사용할 때는 타임존을 고려해서 변환후 사용필요

현재로 가장 좋은 방법은 UTC를 일관되게 사용하는 것으로 보임

dags 폴더에서 코딩시 작성한다면 주의할 점

• Airflow는 dags 폴더를 주기적으로 스캔함

  [core]
  dags_folder = /var/lib/airflow/dags
  # How often (in seconds) to scan the DAGs directory for new files. Default to 5 minutes.
  dag_dir_list_interval = 300

• 이때 DAG 모듈이 들어있는 모든 파일들의 메인 함수가 실행이 됨
• 이 경우 본의 아니게 개발 중인 테스트 코드도 실행될 수 있음

  from airflow import DAG
  …
  cur.execute(“DELETE FROM ….”)

데이터 파이프라인 = ETL = DAG

데이터 웨어하우스의 구성 예

데이터 파이프라인이란?

• ETL: Extract, Transform and Load
• Data Pipeline, ETL, Data Workflow, DAG
  • ETL (Extract, Transform, and Load)
  • Called DAG (Directed Acyclic Graph) in Airflow

ETL vs ELT

• ETL: 데이터를 데이터 웨어하우스 외부에서 내부로 가져오는 프로세스
  • 보통 데이터 엔지니어들이 수행함
• ELT: 데이터 웨어하우스 내부 데이터를 조작해서 (보통은 좀더 추상화되고 요약된) 새로운 데이터를 만드는 프로세스
  • 보통 데이터 분석가들이 많이 수행
  • 이 경우 데이터 레이크 위에서 이런 작업들이 벌어지기도 함
  • 이런 프로세스 전용 기술들이 있으며 dbt가 가장 유명: Analytics Engineering
• • dbt : Data Build Tool **

Data Lake vs. Data Warehouse

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

Data Lake & ELT

Data Pipeline의 정의

• 데이터를 소스로부터 목적지로 복사하는 작업
  • 이 작업은 보통 코딩 (파이썬 혹은 스칼라) 혹은 SQL을 통해 이뤄짐
  • 대부분의 경우 목적지는 데이터 웨어하우스가 됨
• 데이터 소스의 예:
  • Click stream, call data, ads performance data, transactions, sensor data, metadata, …
    • More concrete examples: production databases, log files, API, stream data (Kafka topic)
• 데이터 목적지의 예:
  • 데이터 웨어하우스, 캐시 시스템 (Redis, Memcache), 프로덕션 데이터베이스, NoSQL, S3, ...

데이터 파이프라인의 종류

Raw Data ETL Jobs

1. 외부와 내부 데이터 소스에서 데이터를 읽어다가 (많은 경우 API를 통하게 됨)
2. 적당한 데이터 포맷 변환 후 (데이터의 크기가 커지면 Spark등이 필요해짐)
3. 데이터 웨어하우스 로드

-> 이 작업은 보통 데이터 엔지니어가 함

Summary/Report Jobs

1. DW(혹은 DL)로부터 데이터를 읽어 다시 DW에 쓰는 ETL
2. Raw Data를 읽어서 일종의 리포트 형태나 써머리 형태의 테이블을 다시 만드는 용도
3. 특수한 형태로는 AB 테스트 결과를 분석하는 데이터 파이프라인도 존재

요약 테이블의 경우 SQL (CTAS를 통해)만으로 만들고 이는 데이터 분석가가 하는 것이 맞음. 데이터 엔지니어 관점에서는 어떻게 데이터 분석가들이 편하게 할 수 있는 환경을 만들어 주느냐가 관건 -> Analytics Engineer (DBT)

Production Data Jobs

1. DW로부터 데이터를 읽어 다른 Storage(많은 경우 프로덕션 환경)로 쓰는 ETL a.써머리 정보가 프로덕션 환경에서 성능 이유로 필요한 경우 b.혹은 머신러닝 모델에서 필요한 피쳐들을 미리 계산해두는 경우

2. 이 경우 흔한 타켓 스토리지: a. Cassandra/HBase/DynamoDB와 같은 NoSQL b. MySQL과 같은 관계형 데이터베이스 (OLTP) c. Redis/Memcache와 같은 캐시 d. ElasticSearch와 같은 검색엔진

데이터 파이프라인을 만들 때 고려할 점

이상 혹은 환상

• 내가 만든 데이터 파이프라인은 문제 없이 동작할 것이다
• 내가 만든 데이터 파이프라인을 관리하는 것은 어렵지 않을 것이다

  현실 혹은 실상

• 데이터 파이프라인은 많은 이유로 실패함
  • 버그 :)
  • 데이터 소스상의 이슈: What if data sources are not available or change its data format
  • 데이터 파이프라인들간의 의존도에 이해도 부족
• 데이터 파이프라인의 수가 늘어나면 유지보수 비용이 기하급수적으로 늘어남
  • 데이터 소스간의 의존도가 생기면서 이는 더 복잡해짐. 만일 마케팅 채널 정보가 업데이트가 안된다면 마케팅 관련 다른 모든 정보들이 갱신되지 않음
  • More tables needs to be managed (source of truth, search cost, …)

Best Practices

• 가능하면 데이터가 작을 경우 매번 통채로 복사해서 테이블을 만들기 (Full Refresh)
• Incremental update만이 가능하다면, 대상 데이터소스가 갖춰야할 몇 가지 조건이 있음
  • 데이터 소스가 프로덕션 데이터베이스 테이블이라면 다음 필드가 필요: ■ created (데이터 업데이트 관점에서 필요하지는 않음) ■ modified ■ deleted
  • 데이터 소스가 API라면 특정 날짜를 기준으로 새로 생성되거나 업데이트된 레코드들을 읽어올 수 있어야
• 멱등성(Idempotency)을 보장하는 것이 중요
• 멱등성은 무엇인가?
  • 동일한 입력 데이터로 데이터 파이프라인을 다수 실행해도 최종 테이블의 내용이 달라지지 말아야함 ■ 예를 들면 중복 데이터가 생기지 말아야함
  • 중요한 포인트는 critical point들이 모두 one atomic action으로 실행이 되어야 한다는 점 ■ SQL의 transaction이 꼭 필요한 기술
• 실패한 데이터 파이프라인을 재실행이 쉬어야함
• 과거 데이터를 다시 채우는 과정(Backfill)이 쉬어야 함
• Airflow는 이 부분(특히 backfill)에 강점을 갖고 있음
• 데이터 파이프라인의 입력과 출력을 명확히 하고 문서화
  • 비지니스 오너 명시: 누가 이 데이터를 요청했는지를 기록으로 남길 것!
  • 이게 나중에 데이터 카탈로그로 들어가서 데이터 디스커버리에 사용 가능함 ■ 데이터 리니지가 중요해짐 -> 이걸 이해하지 못하면 온갖 종류의 사고 발생
• 주기적으로 쓸모없는 데이터들을 삭제
  • Kill unused tables and data pipelines proactively
  • Retain only necessary data in DW and move past data to DL (or storage)
• 데이터 파이프라인 사고시 마다 사고 리포트(post-mortem) 쓰기
  • 목적은 동일한 혹은 아주 비슷한 사고가 또 발생하는 것을 막기 위함
  • 사고 원인(root-cause)을 이해하고 이를 방지하기 위한 액션 아이템들의 실행이 중요해짐
  • 기술 부채의 정도를 이야기해주는 바로미터
• 중요 데이터 파이프라인의 입력과 출력을 체크하기
  • 아주 간단하게 입력 레코드의 수와 출력 레코드의 수가 몇개인지 체크하는 것부터 시작
  • 써머리 테이블을 만들어내고 Primary key가 존재한다면 Primary key uniqueness가 보장되는지 체크하는 것이 필요함
  • 중복 레코드 체크

-> 데이터 대상 유닛 테스트

Extract, Transform, Load

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

Airflow 소개

Airflow는 파이썬으로 작성된 데이터 파이프라인 (ETL) 프레임웍

• Airbnb에서 시작한 아파치 오픈소스 프로젝트
• 가장 많이 사용되는 데이터 파이프라인 관리/작성 프레임웍

  데이터 파이프라인 스케줄링 지원

• 정해진 시간에 ETL 실행 혹은 한 ETL의 실행이 끝나면 다음 ETL 실행
• 웹 UI를 제공하기도 함

  데이터 파이프라인(ETL)을 쉽게 만들 수 있도록 해줌

• 다양한 데이터 소스와 데이터 웨어하우스를 쉽게 통합해주는 모듈 제공
• https://airflow.apache.org/docs/
• 데이터 파이프라인 관리 관련 다양한 기능을 제공해줌: 특히 Backfill

  Airflow에서는 데이터 파이프라인을 DAG(Directed Acyclic Graph)라고 부름

• 하나의 DAG는 하나 이상의 태스크로 구성됨

  2020년 12월에 Airflow 2.0이 릴리스됨

  Airflow 버전 선택 방법: 큰 회사에서 사용하는 버전이 무엇인지 확인.

• https://cloud.google.com/composer/docs/concepts/versioning/composer-versions

Airflow 구성

5개의 컴포넌트

1. 웹 서버 (Web Server)
2. 스케줄러 (Scheduler)
3. 워커 (Worker)
4. 메타 데이터 데이터베이스 - Sqlite가 기본으로 설치됨
5. 큐 (다수서버 구성인 경우에만 사용됨) - 이 경우 Executor가 달라짐

• 스케줄러는 DAG들을 워커들에게 배정하는 역할을 수행
• 웹 UI는 스케줄러와 DAG의 실행 상황을 시각화해줌
• 워커는 실제로 DAG를 실행하는 역할을 수행
• 스케줄러와 각 DAG의 실행결과는 별도 DB에 저장됨
  • 기본으로 설치되는 DB는 SQLite
  • 실제 프로덕션에서는 MySQL이나 Postgres를 사용해야함

    Airflow 구조

    

Airflow 스케일링 방법

• 스케일 업 (더 좋은 사양의 서버 사용)
• 스케일 아웃 (서버 추가)

Airflow 구조: 다수 서버

Airflow 구조 다시 보기

여러종류의 Executor들

• Sequential Executor
• Local Executor
• Celery Executor
• Kubernetes Executor
• CeleryKubernetes Executor
• Dask Executor

Airflow 개발의 장단점

장점

• 데이터 파이프라인을 세밀하게 제어 가능
• 다양한 데이터 소스와 데이터 웨어하우스를 지원
• 백필(Backfill)이 쉬움

  단점

• 배우기가 쉽지 않음
• 상대적으로 개발환경을 구성하기가 쉽지 않음
• 직접 운영이 쉽지 않음. 클라우드 버전 사용이 선호됨
  • GCP provides “Cloud Composer”
  • AWS provides “Managed Workflows for Apache Airflow”
  • Azure provides “Data Factory Managed Airflow”

DAG란 무엇인가?

• Directed Acyclic Graph의 줄임말
• Airflow에서 ETL을 부르는 명칭
• DAG는 태스크로 구성됨
  • 예를 3개의 태스크로 구성된다면 Extract, Transform, Load로 구성
• 태스크란? - Airflow의 오퍼레이터(Operator)로 만들어짐
  • Airflow에서 이미 다양한 종류의 오퍼레이터를 제공함
  • 경우에 맞게 사용 오퍼레이터를 결정하거나 필요하다면 직접 개발
  • e.g., Redshift writing, Postgres query, S3 Read/Write, Hive query, Spark job, shell script

Snowflake 소개

• 2014년에 클라우드 기반 데이터웨어하우스로 시작됨 (2020년 상장)
  • 지금은 데이터 클라우드라고 부를 수 있을 정도로 발전
  • 글로벌 클라우드위에서 모두 동작 (AWS, GCP, Azure) - 멀티클라우드
  • 데이터 판매를 통한 매출을 가능하게 해주는 Data Sharing/Marketplace 제공
  • ETL과 다양한 데이터 통합 기능 제공

Snowflake 특징

• 스토리지와 컴퓨팅 인프라가 별도로 설정되는 가변 비용 모델
  • Redshift 고정비용처럼 노드 수를 조정할 필요가 없고 distkey등의 최적화 불필요
• SQL 기반으로 빅데이터 저장, 처리, 분석을 가능하게 해줌
  • 비구조화된 데이터 처리와 머신러닝 기능도 제공
• CSV, JSON, Avro, Parquet 등과 같은 다양한 데이터 포맷을 지원
  • S3, GC 클라우드 스토리지, Azure Blog Storage도 지원
• 배치 데이터 중심이지만 실시간 데이터 처리 지원
• Time Travel: 과거 데이터 쿼리 기능으로 트렌드를 분석하기 쉽게 해줌
• 웹 콘솔 이외에도 Python API를 통한 관리/제어 가능
  • ODBC/JDBC 연결도 지원
• 자체 스토리지 이외에도 클라우드 스토리지를 외부 테이블로 사용 가능
• 대표 고객: Siemens, Flexport, Iterable, Affirm, PepsiCo, …
• 멀티클라우드와 다른 지역에 있는 데이터 공유(Cross-Region Replication) 기능 지원
• Snowflake의 계정 구성도: Organization -> 1+ Account -> 1+ Databases
• Organizations:
  • 한 고객이 사용하는 모든 Snowflake 자원들을 통합하는 최상위 레벨 컨테이너
  • 하나 혹은 그 이상의 Account들로 구성되며 이 모든 Account들의 접근권한, 사용트래킹, 비용들을 관리하는데 사용됨
• Accounts:
  • 하나의 Account는 자체 사용자, 데이터, 접근권한을 독립적으로 가짐
  • 한 Account는 하나 혹은 그 이상의 Database로 구성됨
• Databases:
  • 하나의 Database는 한 Account에 속한 데이터를 다루는 논리적인 컨테이너
  • 한 Database는 다수의 스키마와 거기에 속한 테이블과 뷰등으로 구성되어 있음
  • 하나의 Database는 PB단위까지 스케일 가능하고 독립적인 컴퓨팅 리소스를 갖게 됨 컴퓨팅 리소스를 Warehouses라고 부름. Warehouses와 Databases는 일대일 관계가 아님
• Data Marketplace
  • 데이터 메시 용어가 생기기 전부터 “데이터 마켓플레이스"라는 서비스 제공
• Data Sharing (“Share, Don’t Move”)
  • “Data Sharing”: 데이터 셋을 사내 혹은 파트너에게 스토리지 레벨에서 공유하는 방식

Snowflake의 기본 데이터 타입

Redshift보다 강력한 타입들이 많다 googlecloud의 Bigquery와 비슷

• Numeric: TINYINT, SMALLINT, INTEGER, BIGINT, NUMBER, NUMERIC, DECIMAL, FLOAT, DOUBLE, REAL.
• Boolean: BOOLEAN.
• String: CHAR, VARCHAR, TEXT, BINARY, VARBINARY.
• Date and Time: DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ, TIMESTAMP_TZ.
• Semi-structured data: VARIANT (JSON, OBJECT.
• Binary: BINARY, VARBINARY
• Geospatial: GEOGRAPHY, GEOMETRY.
• Array: ARRAY
• Object: OBJECT

Snowflake Warehouse에서 Credit이란?

• 쿼리 실행과 데이터 로드와 기타 작업 수행에 소비되는 계산 리소스를 측정하는 단위
• 1 credit는 상황에 따라 다르지만 대략 $2-$4의 비용을 발생시킴

  Snowflake 비용 구조

• 크게 아래 3 가지 컴포넌트로 구성됨
• 컴퓨팅 비용: 앞서 크레딧으로 결정됨
• 스토리지 비용: TB 당으로 계산
• 네트워크 비용: 지역간 데이터 전송 혹은 다른 클라우드간 데이터 전송시 TB 당 계산

Snowflake Schema

데이터베이스 밑에 3개의 스키마를 생성

• raw_data
• analytics
• adhoc

Snowflake 사용자 권한 설정

컬럼 레벨 보안 (Column Level Security)

• 테이블내의 특정 컬럼(들)을 특정 사용자나 특정 역할(Role)에만 접근 가능하게 하는 것
• 보통 개인정보 등에 해당하는 컬럼을 권한이 없는 사용자들에게 감추는 목적으로 사용됨
  • 사실 가장 좋은 방법은 아예 그런 컬럼을 별도 테이블로 구성하는 것임
  • 더 좋은 방법은 보안이 필요한 정보를 아예 데이터 시스템으로 로딩하지 않는 것임

    레코드 레벨 보안 (Row Level Security)

• 테이블내의 특정 레코드(들)을 특정 사용자나 특정 역할에만 접근 가능하게 하는 것
• 특정 사용자/그룹의 특정 테이블 대상 SELECT, UPDATE, DELETE 작업에 추가 조건을 다는 형태로 동작
• 일반적으로 더 좋은 방법은 아예 별도의 테이블로 관리하는 것임
  • 다시 한번 더 좋은 방법은 보안이 필요한 정보를 아예 데이터 시스템으로 로딩하지 않는 것임

    Data Governance 관련 기능

• Object Tagging
• Data Classification
• Tag based Masking Policies
• Access History
• Object Dependencies

  잠깐! Data Governance란 무엇인가?

• 필요한 데이터가 적재적소에 올바르게 사용됨을 보장하기 위한 데이터 관리 프로세스
  • 품질 보장과 데이터 관련 법규 준수를 주 목적으로 함
• 다음을 이룩하기 위함이 기본 목적
  • 데이터 기반 결정에서 일관성 예: KPI등의 지표 정의와 계산에 있어 일관성
  • 데이터를 이용한 가치 만들기 Citizen data scientist가 더 효율적으로 일할 수 있게 도와주기 Data silos를 없애기
  • 데이터 관련 법규 준수 개인 정보 보호 -> 적절한 권한 설정과 보안 프로세스 필수!

Data Governance 관련 기능 - Object Tagging

• Enterprise 레벨에서만 가능한 기능. CREATE TAG로 생성
  • 문자열을 Snowflake object에 지정 가능 (계정, 스키마, 테이블, 컬럼, 뷰 등등)
  • 시스템 태그도 있음 (뒤의 Data Classification에서 다시 이야기)
• 이렇게 지정된 tag는 구조를 따라 계승됨

  Data Governance 관련 기능 - Data Classification

• Enterprise 레벨에서만 가능한 기능
• 앞서 Object Tagging은 개인 정보 관리가 주요 용도 중의 하나
  • 하지만 이를 매뉴얼하게 관리하기는 쉽지 않음. 그래서 나온 기능이 Data Classification
• 3가지 스텝으로 구성됨
  • Analyze: 테이블에 적용하면 개인정보나 민감정보가 있는 컬럼들을 분류해냄
  • Review: 이를 사람(보통 데이터 엔지니어)이 보고 최종 리뷰 (결과 수정도 가능)
  • Apply: 이 최종 결과를 System Tag로 적용 SNOWFLAKE.CORE.PRIVACY_CATEGORY (상위레벨)

  • IDENTIFIER, QUASI_IDENTIFIER, SENSITIVE SNOWFLAKE.CORE.SEMANTIC_CATEGORY (하위레벨 - 더 세부정보)

Data Governance 관련 기능 - 식별자와 준식별자

• 개인을 바로 지칭하는 식별자 (Identifier)
• 몇 개의 조합으로 지칭가능한 준식별자 (Quasi Identifier)

Data Governance 관련 기능 - Tag based Masking Policies

• Enterprise 레벨에서만 가능한 기능
  • 먼저 Tag에 액세스 권한을 지정
• 해당 Tag가 지정된 Snowflake Object의 액세스 권한을 그에 맞춰 제한하는 방식
  • 보통 앞서 본 개인정보와 같은 Tag에 부여하는 것이 가장 많이 사용되는 패턴
• Tag Lineage가 여기에도 적용됨.

Data Governance 관련 기능 - Access History

• Enterprise 레벨에서만 가능한 기능
• 목적은 데이터 액세스에 대한 감사 추적을 제공하여 보안과 규정 준수
  • 잠재적인 보안 위반이나 무단 액세스 시도의 조사를 가능하게 해줌
  • 캡처된 정보에는 사용자 신원, IP 주소, 타임스탬프 및 기타 관련 세부 정보 포함
• 'Access History'를 통해 다음 활동의 추적이 가능
  • 데이터베이스 로그인, 실행된 쿼리, 테이블 및 뷰 액세스, 데이터 조작 작업
• 이 기능은 사실 다른 모든 클라우드 데이터 웨어하우스에서도 제공됨

Data Governance 관련 기능 - Object Dependencies

• 데이터 거버넌스와 시스템 무결성 유지를 목적으로 함
• 테이블이나 뷰를 수정하는 경우 이로 인한 영향을 자동으로 식별
  • 예를 들어 테이블 이름이나 컬럼 이름을 변경하거나 삭제하는 경우
  • 즉 데이터 리니지 분석을 자동으로 수행해줌
• 계승 관계 분석을 통한 더 세밀한 보안 및 액세스 제어
  • 어떤 테이블의 개인정보 컬럼이 새로운 테이블을 만들때 사용된다면? 원본 테이블에서의 권한 설정이 그대로 전파됨 (Tag 포함)

사용자별 테이블 권한 설정

• 일반적으로 사용자별 테이블별 권한 설정은 하지 않음
  • 너무 복잡하고 실수의 가능성이 높음
• 역할 (Role) 혹은 그룹(Group) 별로 스키마별 접근 권한을 주는 것이 일반적
  • RBAC(Role Based Access Control)가 새로운 트렌드: 그룹 보다 더 편리
  • 여러 역할에 속한 사용자의 경우는 각 역할의 권한을 모두 갖게 됨 (Inclusive)
• 개인정보와 관련한 테이블들이라면 별도 스키마 설정
  • 극히 일부 사람만 속한 역할에 접근 권한을 줌
• 뒤의 예는 그룹에 적용했지만 GROUP이란 키워드를 ROLE로 바꾸어도 동작

  사용자 그룹 권한 설정

• 앞서 생성한 그룹들의 권한을 아래처럼 설정하고 싶음

analytics_authors

GRANT ALL ON SCHEMA analytics TO GROUP analytics_authors;
GRANT ALL ON ALL TABLES IN SCHEMA analytics TO GROUP analytics_authors;
GRANT ALL ON SCHEMA adhoc to GROUP analytics_authors;
GRANT ALL ON ALL TABLES IN SCHEMA adhoc TO GROUP analytics_authors;
GRANT USAGE ON SCHEMA raw_data TO GROUP analytics_authors;
GRANT SELECT ON ALL TABLES IN SCHEMA raw_data TO GROUP analytics_authors;

analytics_users

GRANT USAGE ON SCHEMA analytics TO GROUP analytics_users;
GRANT SELECT ON ALL TABLES IN SCHEMA analytics TO GROUP analytics_users;
GRANT ALL ON ALL TABLES IN SCHEMA adhoc TO GROUP analytics_users;
GRANT ALL ON SCHEMA adhoc to GROUP analytics_users;
GRANT USAGE ON SCHEMA raw_data TO GROUP analytics_users;
GRANT SELECT ON ALL TABLES IN SCHEMA raw_data TO GROUP analytics_users;

pii_users

GRANT USAGE ON SCHEMA pii TO GROUP pii_users;
GRANT SELECT ON ALL TABLES IN SCHEMA pii TO GROUP pii_users;

컬럼 레벨 보안 (Column Level Security)

• 테이블내의 특정 컬럼(들)을 특정 사용자나 특정 그룹/역할에만 접근 가능하게 하는 것
• 보통 개인정보 등에 해당하는 컬럼을 권한이 없는 사용자들에게 감추는 목적으로 사용됨
  • 사실 가장 좋은 방법은 아예 그런 컬럼을 별도 테이블로 구성하는 것임
  • 더 좋은 방법은 보안이 필요한 정보를 아예 데이터 시스템으로 로딩하지 않는 것임

레코드 레벨 보안 (Row Level Security)

• 테이블내의 특정 레코드(들)을 특정 사용자나 특정 그룹/역할에만 접근 가능하게 하는 것
• 특정 사용자/그룹의 특정 테이블 대상 SELECT, UPDATE, DELETE 작업에 추가 조건을 다는 형태로 동작
  • 이를 RLS (Record Level Security) Policy라고 부름
  • CREATE RLS POLICY 명령을 사용하여 Policy를 만들고 이를 ATTACH RLS POLICY 명령을 사용해 특정 테이블에 추가함
• 일반적으로 더 좋은 방법은 아예 별도의 테이블로 관리하는 것임
  • 다시 한번 더 좋은 방법은 보안이 필요한 정보를 아예 데이터 시스템으로 로딩하지 않는 것임

Redshift 백업과 테이블 복구

Redshift가 지원하는 데이터 백업 방식

• 기본적으로 백업 방식은 마지막 백업으로부터 바뀐 것들만 저장하는 방식
  • 이를 Snapshot이라고 부름
  • 백업을 통해 과거로 돌아가 그 시점의 내용으로 특정 테이블을 복구하는 것이 가능 (Table Restore)
  • 또한 과거 시점의 내용으로 Redshift 클러스터를 새로 생성하는 것도 가능
• 자동 백업:
  • 기본은 하루이지만 최대 과거 35일까지의 변경을 백업하게 할 수 있음.
  • 이 경우 백업은 같은 지역에 있는 S3에 이뤄짐.
  • 다른 지역에 있는 S3에 하려면 Cross-regional snapshot copy를 설정해야함. 이는 보통 재난시 데이터 복구에 유용함
• 매뉴얼 백업:
  • 언제든 원할 때 만드는 백업으로 명시적으로 삭제할 때까지 유지됨 (혹은 생성시 보존 기한 지정)

Redshift Serverless가 지원하는 데이터 백업 방식

• 고정비용 Redshift에 비하면 제한적이고 조금더 복잡함
• 일단 Snapshot 이전에 Recovery Points라는 것이 존재
  • Recovery Point를 Snapshot으로 바꾼 다음에 여기서 테이블 복구를 하거나 이것으로 새로운 Redshift 클러스터 등을 생성하는 것이 가능
• Recovery Points는 과거 24시간에 대해서만 유지됨

Redshift 관련 기타 서비스 소개

Redshift Spectrum

• Redshift의 확장 기능
• S3에 있는 파일들을 마치 테이블처럼 SQL로 처리 가능
  • S3 파일들을 외부 테이블들(external table)로 처리하면서 Redshift 테이블과 조인 가능
  • S3 외부 테이블들은 보통 Fact 테이블들이 되고 Redshift 테이블들은 Dimension 테이블
  • 1TB를 스캔할 때마다 $5 비용이 생김
• 이를 사용하려면 Redshift 클러스터가 필요
  • S3와 Redshift 클러스터는 같은 region에 있어야함

Redshift Serverless

• Redshift의 경우 용량을 미리 결정하고 월정액 (Fixed Cost) 지급
• Redshift Serverless는 반대로 쓴만큼 비용을 지불하는 옵션
  • BigQuery와 같은 사용한 자원에 따른 비용 산정 방식
  • 데이터 처리 크기와 특성에 따라 오토 스케일링이 적용됨

Athena

• AWS의 Presto 서비스로 사실상 Redshift Spectrum과 비슷한 기능을 제공
• S3에 있는 데이터들을 기반으로 SQL 쿼리 기능 제공
  • 이 경우 S3를 데이터 레이크라 볼 수 있음

Redshift ML

• SQL만 사용하여 머신러닝 모델을 훈련하고 사용할 수 있게 해주는 Redshift 기능
• 이 기능은 사실 AWS SageMaker에 의해 지원됨
  • SageMaker는 Auto Pilot이라 하여 최적화된 모델을 자동 생성해주는 기능 제공
• 이미 모델이 만들어져 있다면 이를 사용하는 것도 가능 (BYOM: Bring Your Own Model)

Redshift Spectrum으로 S3 외부 테이블 조작해보기

Fact 테이블과 Dimension 테이블

• Fact 테이블: 분석의 초점이 되는 양적 정보를 포함하는 중앙 테이블
  • 일반적으로 매출 수익, 판매량 또는 이익과 같은 사실 또는 측정 항목을 포함하며 비즈니스 결정에 사용
  • Fact 테이블은 일반적으로 외래 키를 통해 여러 Dimension 테이블과 연결됨
  • 보통 Fact 테이블의 크기가 훨씬 더 큼
• Dimension 테이블: Fact 테이블에 대한 상세 정보를 제공하는 테이블
  • 고객, 제품과 같은 테이블로 Fact 테이블에 대한 상세 정보 제공
  • Fact 테이블의 데이터에 맥락을 제공하여 사용자가 다양한 방식으로 데이터를 조각내고 분석 가능하게 해줌
  • Dimension 테이블은 일반적으로 primary key를 가지며, fact 테이블의 foreign key에서 참조
  • 보통 Dimension 테이블의 크기는 훨씬 더 작음

Redshift Spectrum 사용 유스 케이스

• S3에 대용량 Fact 테이블이 파일(들)로 존재
• Redshift에 소규모 Dimension 테이블이 존재
• Fact 테이블을 Redshift로 적재하지 않고 위의 두 테이블을 조인하고 싶다면?
• 이 때 사용할 수 있는 것이 Redshift Spectrum
  • 이는 별도로 설정하거나 론치하는 것이 아니라 Redshift의 확장 기능으로 사용하고 그만큼 비용 부담

외부 테이블(External Table)이란?

• 데이터베이스 엔진이 외부에 저장된 데이터를 마치 내부 테이블처럼 사용하는 방법
  • 외부 테이블은 외부(보통 S3와 같은 클라우드 스토리지)에 저장된 대량의 데이터를 데이터베이스 내부로 복사하고 쓰는 것이 아니라 임시 목적으로 사용하는 방식
• SQL 명령어로 데이터베이스에 외부 테이블 생성 가능
  • 이 경우 데이터를 새로 만들거나 하는 것이 아니라 참조만 하게 됨
  • 외부 테이블은 CSV, JSON, XML과 같은 파일 형식 뿐만 아니라 ODBC 또는 JDBC 드라이버를 통해 액세스하는 원격 데이터베이스와 같은 다양한 데이터 소스에 대해 사용 가능
• 외부 테이블을 사용하여 데이터 처리 후 결과를 데이터베이스에 적재하는데 사용가능
  • 예를 들어, 외부 테이블을 사용하여 로그 파일을 읽고 정제된 내용을 데이터베이스 테이블에 적재 가능
• 외부 테이블은 보안 및 성능 문제에 대해 신중한 고려가 필요
• 이는 Hive등에서 처음 시작한 개념으로 이제는 대부분의 빅 데이터 시스템에서 사용됨

Redshift Spectrum 사용 방식

• S3에 있는 파일들을 마치 테이블처럼 SQL로 처리 가능
  • S3 파일들을 외부 테이블들(external table)로 처리하면서 Redshift 테이블과 조인 가능
  • S3 외부 테이블들은 보통 Fact 테이블들이 되고 Redshift 테이블들은 Dimension 테이블
• 이를 사용하려면 Redshift 클러스터가 필요
  • S3와 Redshift 클러스터는 같은 region에 있어야함
• S3 Fact 데이터를 외부 테이블(External Table)로 정의해야함

Redshift Spectrum 실습을 위한 외부 테이블 용 스키마 설정

• 먼저 앞서 만든 redshift.read.s3 ROLE에 AWSGlueConsoleFullAccess 권한 지정이 필요
• 다음으로 아래 SQL을 실행하여 외부 테이블용 스키마 생성

  CREATE EXTERNAL SCHEMA external_schema
  from data catalog
  database 'myspectrum_db'
  iam_role 'arn:aws:iam::521227329883:role/redshift.read.s3'
  create external database if not exists;

잠깐 AWS Glue란 무엇인가?

AWS Glue는 AWS의 Serverless ETL 서비스로 아래와 같은 기능 제공

• 데이터 카탈로그: a. AWS Glue Data Catalog는 데이터 소스 및 대상의 메타데이터를 대상으로 검색 기능을 제공. 이는 주로 S3나 다른 AWS 서비스 상의 데이터 소스를 대상으로 함 (Redshift Spectrum의 경우에는 외부 테이블들)
• ETL 작업 생성: AWS Glue Studio a. 간단한 드래그 앤 드롭 인터페이스를 통해 ETL 작업 생성 가능 b. 사용자는 데이터 소스 및 대상을 선택하고 데이터 변환 단계를 정의하는 스크립트 생성
• 작업 모니터링 및 로그: a. AWS Glue 콘솔을 통해 사용자는 ETL 작업의 실행 상태 및 로그를 모니터링 가능
• 서버리스 실행: a. AWS Glue는 서버리스 아키텍처를 사용하므로 사용자는 작업을 실행하는 데 필요한 인프라를 관리할 필요가 없음 (Auto Scaling)

Redshift ML 사용하기

머신러닝의 정의

• 배움이 가능한 기계(혹은 알고리즘)의 개발
  • 결국 데이터의 패턴을 보고 흉내(imitation)내는 방식으로 학습
  • 학습에 사용되는 이 데이터를 트레이닝셋 (training set)이라고 부름
• 컴퓨터가 학습할 수 있도록 하는 알고리즘과 기술을 개발하는 분야
• 딥러닝(신경망의 다른 이름)은 머신 러닝의 일부
  • 비젼, 자연언어처리 (텍스트/오디오)등에 적용되고 있음
• 인공지능은 머신러닝을 포괄하는 개념

머신러닝 모델이란?

머신 러닝의 최종 산물이 머신 러닝 모델

• 학습된 패턴(트레이닝셋)에 따라 예측을 해주는 블랙박스
  • 선택한 머신러닝 학습 알고리즘에 따라 내부가 달라짐
  • 디버깅은 쉽지 않으며 왜 동작하는지 이유를 설명하기도 쉽지 않음
  • 트레이닝셋의 품질이 머신러닝 모델의 품질을 결정

    입력 데이터를 주면 그를 기반으로 예측

• 정확히 이야기하자면 지도 머신러닝 (Supervised Machine Learning)
• 이외에도 2가지의 다른 머신러닝 방식이 존재
  • 비지도 머신러닝(Unsupervised Machine Learning)
  • 강화 학습 (Reinforcement Learning)

    머신러닝 모델 트레이닝 혹은 빌딩이란?

• 이런 머신 러닝 모델을 만드는 것을 지칭
• 입력은 트레이닝셋

Amazon SageMaker란?

머신러닝 모델 개발을 처음부터 끝까지 해결해주는 AWS 서비스

• MLOps 프레임웍

  크게 4가지 기능 제공

• 트레이닝 셋 준비
• 모델 훈련
• 모델 검증
• 모델 배포와 관리
  • API 엔드포인트, 배치 서빙, …

    다양한 머신러닝 프레임웍을 지원

• Tensorflow/Keras, PyTorch, MXNet, …
• 자체 SageMaker 모듈로 머신러닝 모델 훈련 가능

  SageMaker Studio라는 웹기반 환경 제공 (노트북)

  다양한 개발방식 지원

• 기본적으로 Python Notebook (SageMaker 모듈)을 통해 모델 훈련
  • 스칼라/자바 SDK도 제공
• AutoPilot이라는 코딩 불필요 모델 훈련 기능 제공
  • 이 경우에도 코드를 만들어줌

    다른 클라우드 업체들도 비슷한 프레임웍 제공

SageMaker의 AutoPilot 소개

• AutoPilot: SageMaker에서 제공되는 AutoML 기능
  • AutoML이란 모델빌딩을 위한 훈련용 데이터 셋을 제공하면 자동으로 모델을 만들어주는 기능
• AutoPilot은 훈련용 데이터 셋을 입력으로 다음을 자동으로 수행
  • 먼저 데이터 분석(EDA: Exploratory Data Analysis)을 수행하고 이를 파이썬 노트북으로 만들어줌
  • 다수의 머신 러닝 알고리즘과 하이퍼 파라미터의 조합에 대해 아래 작업을 수행 머신 러닝 모델을 만들고 훈련하고 테스트하고 테스트 결과를 기록
  • 선택 옵션에 따라 모델 테스트까지 다 수행하기도 하지만 코드를 만드는 단계(노트북)로 마무리도 가능 즉, AutoPilot 기능을 통해 모델개발 속도를 단축하는 것이 가능
• 최종적으로 사용자가 모델을 선택 후 API로 만드는 것도 가능
  • 여기에 로그를 설정할 수 있음 (전체 로깅이나 샘플 로깅 설정 가능)

Redshift 중지/제거하기

Redshift 관련 유지보수

• Redshift 서비스는 주기적으로 버전 업그레이드를 위해 중단됨
  • 이를 Maintenance window라고 부름
  • Serverless에는 이게 존재하지 않음

    테이블 청소와 최적화 - VACUUM 명령

• 테이블 데이터 정렬:
  • Redshift 테이블에 데이터가 삽입, 업데이트 또는 삭제될 때 데이터는 불규칙하게 분산되어 저장될 수 있는데 VACUUM 명령어는 데이터를 정렬하여 남아 있는 행을 모아 쿼리 실행 시 검색해야 할 블록 수를 줄이는 작업 수행
• 디스크 공간 해제:
  • 테이블에서 행이 삭제되면 디스크 공간이 즉시 해제되지 않음.
  • VACUUM 명령어는 더 이상 필요하지 않은 행을 제거하고 사용한 디스크 공간을 해제
• 삭제된 행에서 공간 회수:
  • 테이블에서 행이 삭제되면 VACUUM 명령 실행 전까지 이 공간은 회수되지 않음
• 테이블 통계 업데이트:
  • VACUUM은 테이블 통계를 업데이트하여 Query Planner가 쿼리 최적화 지원
• 큰 테이블에 대한 VACUUM 명령은 리소스를 많이 잡아먹음
  • 바쁘지 않을 때 실행해주는 것이 좋음

(고정 비용) Redshift 클러스터 중지/재실행

• Redshift가 당분간 필요없다면?
  • Redshift 콘솔에서 해당 Redshift 클러스터를 선택하고 상단 메뉴에서 Stop 선택
  • 이 경우 Redshift 클러스터의 스토리지 비용만 부담. 당연히 SQL 실행은 불가능
• Redshift가 다시 필요해지면
  • 같은 메뉴에서 Resume 선택

(고정 비용) Redshift 클러스터 삭제

• Redshift가 영원히 필요없다면?
  • Redshift 콘솔에서 삭제할 클러스터를 선택하고 상단 메뉴에서 Delete 선택
  • 이 때 데이터베이스 내용 백업을 S3로 할지 여부를 선택 가능
  • 이 S3 백업으로부터 Redshift 클러스터를 나중에 새로 론치 가능함

(가변 비용) Redshift Serverless 삭제

• 먼저 모든 Workgroup들을 삭제
• 다음으로 모든 Namespace들을 삭제

• AWS에서 지원하는 데이터 웨어하우스 서비스
• 2 PB의 데이타까지 처리 가능
  • 최소 160GB로 시작해서 점진적으로 용량 증감 가능
• Still OLAP
  • 응답속도가 빠르지 않기 때문에 프로덕션 데이터베이스로 사용불가
• 컬럼 기반 스토리지
  • 레코드 별로 저장하는 것이 아니라 컬럼별로 저장함
  • 컬럼별 압축이 가능하며 컬럼을 추가하거나 삭제하는 것이 아주 빠름
• 벌크 업데이트 지원
  • 레코드가 들어있는 파일을 S3로 복사 후 COPY 커맨드로 Redshift로 일괄 복사
• 고정 용량/비용 SQL 엔진
  • 최근 가변 비용 옵션도 제공 (Redshift Serverless)
• 데이터 공유 기능 (Datashare):
  • 다른 AWS 계정과 특정 데이터 공유 가능. Snowflake의 기능을 따라함
• 다른 데이터 웨어하우스처럼 primary key uniqueness를 보장하지 않음
  • 프로덕션 데이터베이스들은 보장함

Redshift는 SQL 기반 관계형 데이터베이스

• Postgresql 8.x와 SQL이 호환됨
  • 하지만 Postgresql 8.x의 모든 기능을 지원하지는 않음
  • 예를 들어 text 타입이 존재하지 않음
• Postgresql 8.x를 지원하는 툴이나 라이브러리로 액세스 가능
  • JDBC/ODBC
• 다시 한번 SQL이 메인 언어라는 점 명심
  • 그래서 데이터 모델링(테이블 디자인)이 아주 중요

Redshift의 스케일링 방식

• 용량이 부족해질 때마다 새로운 노드를 추가하는 방식으로 스케일링
• 예) Scale Out 방식과 Scale Up 방식
  • dc2.large가 하나면 최대 0.16TB까지의 용량을 갖게됨
  • 공간이 부족해지면

  • dc2.large 한대를 더 추가 -> 총 0.32TB (Scale Out)

  • 아니면 사양을 더 좋은 것으로 업그레이드 -> dc2.8xlarge 한대로 교체 (Scale Up)

• 이를 Resizing이라 부르며 Auto Scaling 옵션을 설정하면 자동으로 이뤄짐
• 이는 Snowflake나 BigQuery의 방식과는 굉장히 다름
  • 여기서는 특별히 용량이 정해져있지 않고 쿼리를 처리하기 위해 사용한 리소스에 해당하는 비용 지불

  • 즉 Snowflake와 BigQuery가 훨씬 더 스케일하는 데이터베이스 기술이라 볼 수 있음

  • 장단점 존재 -> 비용의 예측이 불가능하다는 단점 존재

• Redshift에도 가변비용 옵션 존재 -> Redshift Serverless
  • 뒤에 데모에서는 Redshift Serverless를 사용해볼 예정 (Pay as You Go)

Redshift 최적화는 굉장히 복잡

• Redshift가 두 대 이상의 노드로 구성되면 한 테이블의 레코드들의 저장방식은?
  • 분산 저장되어야함
  • 또 한 노드 내에서는 순서가 정해주어야함

Redshift의 레코드 분배와 저장 방식

Redshift가 두 대 이상의 노드로 구성되면 그 시점부터 테이블 최적화가 중요

• 한 테이블의 레코드들을 어떻게 다수의 노드로 분배할 것이냐!

  Distkey, Diststyle, Sortkey 세 개의 키워드를 알아야함

• Diststyle은 레코드 분배가 어떻게 이뤄지는지를 결정
  • all, even, key (디폴트는 “even”)
• Distkey는 레코드가 어떤 컬럼을 기준으로 배포되는지 나타냄 (diststyle이 key인 경우)
• Sortkey는 레코드가 한 노드내에서 어떤 컬럼을 기준으로 정렬되는지 나타냄
  • 이는 보통 타임스탬프 필드가 됨

    Diststyle이 key인 경우 컬럼 선택이 잘못되면?

• 레코드 분포에 Skew가 발생 -> 분산처리의 효율성이 사라짐
• BigQuery나 Snowflake에서는 이런 속성을 개발자가 지정할 필요가 없음 (시스템이 알아서 선택)

Redshift의 레코드 분배와 저장 방식 예

CREATE TABLE my_table (
 column1 INT,
 column2 VARCHAR(50),
 column3 TIMESTAMP,
 column4 DECIMAL(18,2)
) DISTSTYLE KEY DISTKEY(column1) SORTKEY(column3);
-- my_table의 레코드들은 column1의 값을 기준으로 분배되고 같은 노드(슬라이스)안에서는 column3의 값을 기준으로 소팅이 됨

Redshift의 벌크 업데이트 방식 - COPY SQL

Redshift의 기본 데이터 타입

• SMALLINT (INT2) • INTEGER (INT, INT4) • BIGINT (INT8) • DECIMAL (NUMERIC) • REAL (FLOAT4) • DOUBLE PRECISION (FLOAT8) • BOOLEAN (BOOL) • CHAR (CHARACTER) • VARCHAR (CHARACTER VARYING) • TEXT (VARCHAR(256)) • DATE • TIMESTAMP

Redshift의 고급 데이터 타입

• GEOMETRY • GEOGRAPHY • HLLSKETCH • SUPER

Redshift 초기 설정

Redshift Schema: 다른 기타 관계형 데이터베이스와 동일한 구조

스키마(Schema) 설정

• 모든 스키마를 리스트하기: select * from pg_namespace;

  CREATE SCHEMA raw_data;
  CREATE SCHEMA analytics;
  CREATE SCHEMA adhoc;
  CREATE SCHEMA pii;

  사용자(User) 생성

• 모든 사용자를 리스트하기: select * from pg_user;

  CREATE USER keeyong PASSWORD '...';

  그룹(Group) 생성/설정

• 한 사용자는 다수의 그룹에 속할 수 있음

• 그룹의 문제는 계승이 안된다는 점

  • 즉 너무 많은 그룹을 만들게 되고 관리가 힘들어짐

• 예를 들어 다음과 같은 그룹이 존재

  • 어드민을 위한 pii_users
  • 데이터 분석가를 위한 analytics_authors
  • 데이터 활용을 하는 개인을 위한 analytics_users

• 그룹 생성 - CREATE GROUP

• 그룹에 사용자 추가 - ALTER GROUP 그룹이름 ADD USER 사용자이름

• 그룹에 스키마/테이블 접근 권한 설정 (나중에 설명)

• 모든 그룹을 리스트하기: select * from pg_group;

  CREATE GROUP analytics_users;
  CREATE GROUP analytics_authors;
  CREATE GROUP pii_users;
  ALTER GROUP analytics_authors ADD USER keeyong;
  ALTER GROUP analytics_users ADD USER keeyong;
  ALTER GROUP pii_users ADD USER keeyong;

  역할(Role) 생성/설정

• 역할은 그룹과 달리 계승 구조를 만들 수 있음

• 역할은 사용자에게 부여될 수도 있고 다른 역할에 부여될 수도 있음

• 한 사용자는 다수의 역할에 소속가능함

• 모든 역할을 리스트하기: select * from SVV_ROLES;

  CREATE ROLE staff;
  CREATE ROLE manager;
  CREATE ROLE external;
  GRANT ROLE staff TO keeyong; 
  GRANT ROLE staff TO ROLE manager;

데이터 조직의 비전은?

• 신뢰할 수 있는 데이터를 바탕으로 부가 가치 생성
  • Data is the new oil? (데이터가 새로운 석유다)
  • 데이터의 중요성을 강조하니 데이터 팀도 회사에서 인정을 받는다? (매출에 기여를 해야 인정을 받음)

데이터 조직이 하는 일

• 고품질 데이터를 기반으로 의사 결정권자에게 입력 제공 (데이터 분석가)

  • 결정 과학 (Decision Science)라고 부르기도 함.
  • 데이터를 고려한 결정(data informed decisions)을 가능하게 해줌 vs. 데이터 기반 결정(data driven decisions)
  • 예를 들면 데이터 기반 지표 정의, 대시보드와 리포트 생성 등을 수행

• 고품질 데이터를 기반으로 사용자 서비스 경험 개선 혹은 프로세스 최적화 (Product Science)

  • 머신 러닝과 같은 알고리즘을 통해 사용자의 서비스 경험을 개선 예) 개인화를 바탕으로한 추천과 검색 기능 제공
  • 공장이라면 공정 과정에서 오류를 최소화하는 일을 수행

데이터의 흐름과 데이터 팀의 발전 단계

데이터 팀의 발전 - 1. 데이터 인프라 구축

데이터 인프라의 구축은 데이터 엔지니어가 수행함

프로덕션 데이터베이스 vs. 데이터 웨어하우스

데이터 웨어하우스

• 회사에 필요한 모든 데이터를 모아놓은 중앙 데이터베이스 (SQL 데이터베이스)

  • 데이터의 크기에 맞게 어떤 데이터베이스를 사용할지 선택
  • 크기가 커진다면 다음 중 하나를 선택

    ▪ AWS Redshift, 구글 클라우드의 BigQuery ▪ 스노우플레이크(Snowflake) ▪ 오픈소스 기반의 하둡(Hive/Presto)/스팍 ▪ 이 모두 SQL을 지원

• 중요 포인트는 프로덕션용 데이터베이스와 별개의 데이터베이스여야 한다는 점

• 데이터 웨어하우스의 구축이 진정한 데이터 조직이 되는 첫 번째 스텝

ETL(Extract, Transform, Load)이란?

• 다른 곳에 존재하는 데이터를 가져다가 데이터 웨어하우스에 로드하는 작업
  • Extract: 외부 데이터 소스에서 데이터를 추출
  • Transform: 데이터의 포맷을 원하는 형태로 변환
  • Load: 변환된 데이터를 최종적으로 데이터 웨어하우스로 적재
  • 데이터 파이프라인이라고 부르기도 함
• 관련하여 가장 많이 쓰이는 프레임웍은 Airflow
  • Airflow는 오픈소스 프로젝트로 파이썬 3 기반이며 Airbnb에서 시작
  • AWS와 구글 클라우드에서도 지원
• ETL 관련 SaaS (Software as a Service)도 출현하기 시작
  • 흔한 데이터 소스의 경우 FiveTran, Stitch Data와 같은 SaaS를 사용하는 것도 가능

데이터 팀의 발전 - 2. 데이터 분석 수행

이는 데이터 분석가 (Data Analyst)가 맡는 일임

시각화 대시보드란?

• 보통 중요한 지표를 시간의 흐름과 함께 보여주는 것이 일반적
  • 지표의 경우 3A(Accessible, Actionable, Auditable)가 중요
  • 중요 지표의 예: 매출액, 월간/주간 액티브 사용자수, ...
• 가장 널리 사용되는 대시보드:
  • 구글 클라우드의 룩커(Looker)
  • 세일즈포스의 태블로 (Tableau)
  • 마이크로소프트의 파워 BI(Power BI)
  • 오픈소스 아파치 수퍼셋(Superset)

데이터 팀의 발전 - 3. 데이터 과학 적용

머신 러닝(Machine Learning)이란?

• (프로그래밍 없이) 배움이 가능한 알고리즘 -> 블랙박스
  • A field of study that gives computers the ability to learn without being explicitly programmed’ (Arthur Samuel)
• 데이터로부터 패턴을 찾아 학습
  • 데이터의 품질과 크기가 중요
  • 데이터로 인한 왜곡 (bias) 발생 가능 AI 윤리
  • 내부동작 설명 가능 여부도 중요 ML Explainability

데이터 조직의 구성원

데이터 팀에는 누가 있는가?

• 작은 회사에서는 한 사람이 몇 개의 역할을 동시 수행하는 것이 일반적
• 데이터 엔지니어 (Data Engineer)
  • 데이터 인프라 (데이터 웨어하우스와 ETL) 구축
• 데이터 분석가 (Data Analyst)
  • 데이터 웨어하우스의 데이터를 기반으로 지표를 만들고 시각화 (대시보드)
  • 내부 직원들의 데이터 관련 질문 응답
• 데이터 과학자 (Data Scientist)
  • 과거 데이터를 기반으로 미래를 예측하는 머신러닝 모델을 만들어 고객들의 서비스 경험을 개선 (개인화 혹은 자동화 혹은 최적화)

데이터 엔지니어의 역할

• 기본적으로는 소프트웨어 엔지니어
  • 파이썬이 대세. 자바 혹은 스칼라와 같은 언어도 아는 것이 좋음
• 데이터 웨어하우스 구축
  • 데이터 웨어하우스를 만들고 이를 관리. 클라우드로 가는 것이 추세 AWS의 Redshift, 구글클라우드의 BigQuery, 스노우플레이크
  • 관련해서 중요한 작업중의 하나는 ETL 코드를 작성하고 주기적으로 실행해주는 것 ETL 스케줄러 혹은 프레임웍이 필요 (Airflow라는 오픈소스가 대세)
• 데이터 분석가와 과학자 지원
  • 데이터 분석가, 데이터 과학자들과의 협업을 통해 필요한 툴이나 데이터를 제공해주는 것이 데이터 엔지니어의 중요한 역할 중의 하나

데이터 엔지니어가 알아야하는 기술

• SQL: 기본 SQL, Hive, Presto, SparkSQL, …
• 프로그래밍 언어: 파이썬, 스칼라, 자바
• 데이터 웨어하우스
  • Redshift/Snowflake/BigQuery
• ETL/ELT 프레임웍: Airflow, …
• 대용량 데이터 처리 플랫폼: Spark/YARN
• 컨테이너 기술 - Docker/K8s
• 클라우드 컴퓨팅
  • AWS, GCP, Azure
• 도움이 되는 기타 지식
  • 머신 러닝 일반
  • A/B 테스트, 통계
• 데이터 엔지니어 스킬 로드맵
  • https://github.com/datastacktv/data-engineer-roadmap
  • MLOps 혹은 ML Engineer가 다음 스텝이 많이 됨

데이터 분석가의 역할

비지니스 인텔리전스를 책임짐 (의사결정을 객관적이고 과학적이도록)

• 중요 지표를 정의하고 이를 대시보드 형태로 시각화
  • 대시보드로는 태블로(Tableau)와 룩커(Looker)등의 툴이 가장 흔히 사용됨
  • 오픈소스로는 수퍼셋(Superset)이 많이 사용됨
• 이런 일을 수행하려면 비지니스 도메인에 대한 깊은 지식이 필요

  회사내 다른 팀들의 데이터 관련 질문 대답

• 임원들이나 팀 리드들이 데이터 기반 결정을 내릴 수 있도록 도와줌
• 질문들이 굉장히 많고 반복적이기에 어떻게 셀프서비스로 만들 수 있느냐가 관건

데이터 분석가가 알아야하는 기술

• SQL: 기본 SQL, Hive, Presto, SparkSQL, …
• 대시보드
  • 룩커, 태블로, 파워 BI, 수퍼셋
  • 엑셀, 구글 스프레드시트, 파이썬
• 데이터 모델링
• 통계 지식
  • AB 테스트 분석 혹은 다양한 데이터 분석에서 통계 지식은 아주 유용함
• 비지니스 도메인에 관한 깊은 지식
• 좋은 지표를 정의하는 능력
• 보통 코딩을 하지는 않음 (추세는 데이터 엔지니어링 적인 지식이 있으면 좋음, DBT)

데이터 분석가의 딜레마

• 보통 많은 수의 긴급한 데이터 관련 질문들에 시달림
• 좋은 데이터 인프라 없이는 일을 잘 하기 힘듬!
• 많은 경우 현업팀에 소속되기도 함
  • 내 커리어에서 다음은 무엇인가?
  • 소속감이 불분명하고 내 고과 기준이 불명확해짐
• 데이터 분석가의 경우 조직 구조가 더 중요함

어떤 새로운 직군들 혹은 뜨는 서비스들이 있는가?

• ML 엔지니어 (vs. 데이터 과학자 & 데이터 엔지니어)
• ML옵스 (MLOps)
• 프라이버시 엔지니어: 개인정보 보호
• 데이터 디스커버리 서비스

MLOps란 무슨 일을 하는가?

DevOps가 하는 일은?

• 개발자가 만든 코드를 시스템에 반영하는 프로세스 (CI/CD, deployment)
• 시스템이 제대로 동작하는지 모니터링 그리고 이슈 감지시 escalation 프로세스
  • On-call 프로세스

    MLOps가 하는 일은?

• 앞의 DevOps가 하는 일과 동일. 차이점은 서비스 코드가 아니라 ML 모델이 대상
• 모델을 계속적으로 빌딩하고 배포하고 성능을 모니터링
  • ML모델 빌딩과 프로덕션 배포를 자동화할 수 있을까? 계속적인 모델 빌딩(CT)과 배포!
  • 모델 서빙 환경과 모델의 성능 저하를 모니터링하고 필요시 escalation 프로세스 진행

MLOps 엔지니어가 알아야하는 기술

데이터 엔지니어가 알아야 하는 기술

• 파이썬/스칼라/자바
• 데이터 파이프라인과 데이터 웨어하우스

  DevOps 엔지니어가 알아야 하는 기술

• CI/CD, 서비스 모니터링, …
• 컨테이너 기술 (K8S, 도커)
• 클라우드 (AWS, GCP, Azure)

  머신러닝 관련 경험/지식

• 머신러닝 모델 빌딩과 배포
• ML 모델 빌딩 프레임웍 경험
  • SageMaker, Kubeflow, MLflow

프라이버시 엔지니어

전체 시스템에서 개인정보 보호를 위한 가이드라인/툴을 제공

• 개인정보란? 개인을 식별할 수 있는 정보

  이는 데이터 시스템에서 더욱 중요

  개인 정보 보호 법안의 징벌 조항이 점점 강화되는 추세

• 정보 주체의 권리를 강화하는 방향으로도 변화: GDPR의 프로파일링 거부권
• 유럽 연합의 GDPR (General Data Protection Regulation)
• 미국의 HIPAA (건강보험 이전 및 책임에 관한 법률)
• 미국 캘리포니아의 CCPR (캘리포니아 소비자 개인정보 보호 법안)

데이터 디스커버리 (Data Discovery)란?

• 별도 직군은 아니지만 데이터 팀이 커지면 꼭 필요한 서비스
• 데이터가 커지면 테이블과 대시보드의 수도 증가!
  • 데이터 분석시 어느 테이블이나 대시보드를 봐야하는지 혼란이 생김

  • 그러면 에라 모르겠다 직접 새로운 테이블이나 대시보드를 또 만들어냄

  • 정보 과잉 문제가 더 심해지는 악순환!

  • 주기적인 테이블과 대시보드 클린업이 필수!
• 테이블과 대시보드 관련 검색 서비스!
  • 리프트에서 만든 아문센
  • 링크드인에서 만든 데이터허브
  • 셀렉트스타

데이터 웨어하우스와 데이터 레이크와 ETL/ELT

데이터 웨어하우스 옵션별 장단점

• 데이터 웨어하우스는 기본적으로 클라우드가 대세
• 데이터가 커져도 문제가 없는 확장가능성(Scalable)과 적정한 비용이 중요한 포인트
• 크게 고정비용 옵션과 가변비용 옵션이 존재하며 후자가 좀더 확장가능한 옵션
• AWS의 Redshift, 구글 클라우드의 BigQuery, 스노우플레이크(Snowflake)
  • Redshift는 고정비용 옵션이며 BigQuery와 스노우플레이크는 가변비용
• 오픈소스 기반(Presto, Hive)을 사용하는 경우도 클라우드 버전 존재
• 데이터가 작다면 굳이 빅데이터 기반 데이터베이스를 사용할 필요가 없음

데이터 레이크

• 구조화 데이터 + 비구조화 데이터 (로그 파일)
• 보존 기한이 없는 모든 데이터를 원래 형태대로 보존하는 스토리지에 가까움
• 보통은 데이터 웨어하우스보다 몇 배는 더 크고 더 경제적인 스토리지
• 보통 클라우드 스토리지가 됨
  • AWS라면 S3가 대표적인 데이터 레이크라 볼 수 있음
• 데이터 레이크가 있는 환경에서 ETL과 ELT
  • 데이터 레이크와 데이터 웨어하우스 바깥에서 안으로 데이터를 가져오는 것: ETL
  • 데이터 레이크와 데이터 웨어하우스 안에 있는 데이터를 처리하는 것: ELT

ETL(Extract, Transform, Load) -> ELT

ETL의 수는 회사의 성장에 따라 쉽게 100+개 이상으로 발전

• 중요한 데이터를 다루는 ETL이 실패했을 경우 이를 빨리 고쳐서 다시 실행하는 것이 중요
• 이를 적절하게 스케줄하고 관리하는 것이 중요해지며 그래서 ETL 스케줄러 혹은 프레임웍이 필요해짐
  • Airflow가 대표적인 프레임웍

    데이터 요약를 위한 ETL도 필요해짐 -> ELT라고 부름

• 앞에서 설명한 ETL은 다양한 데이터 소스에 있는 데이터를 읽어오는 일을 수행.
• 하지만 이를 모두 이해해서 조인해서 사용하는 것은 데이터가 다양해지고 커지면서 거의 불가능해짐.
• 주기적으로 요약 데이터를 만들어 사용하는 것이 더 효율적. dbt 사용
  • 예) 고객 매출 요약 테이블, 제품 매출 요약 테이블, …

다양한 데이터 소스의 예

• 프로덕션 데이터베이스(웹/앱에서 사용하는 데이터베이스)의 데이터
  • 보통 MySQL, Postgres등이 프로덕션 데이터베이스로 사용됨
• 이메일 마케팅 데이터
  • Mailchimp, HubSpot, SendGrid, ...
• 크레딧카드 매출 데이터
  • Stripe
• 서포트 티켓 데이터
  • Zendesk, Kustomer, ...
• 서포트 콜 데이터
  • ChannelTalk, RingCentral, Talkdesk, …
• 세일즈 데이터
  • Salesforce
• 사용자 이벤트 로그
  • Amplitude, MixPanel, 웹서버로그, ...

Airflow (ETL 스케줄러) 소개

• ETL 관리 및 운영 프레임웍의 필요성
  • 다수의 ETL이 존재할 경우 이를 스케줄해주고 이들간의 의존관계(dependency)를 정의해주는기능 필요
  • 특정 ETL이 실패할 경우 이에 관한 에러 메세지를 받고 재실행해주는 기능도 중요해짐 (Backfill - 데이터 엔지니어한테는 악몽)
• 가장 많이 사용되는 프레임웍은 Airflow
  • Airflow는 오픈소스 프로젝트로 파이썬 3 기반이며 에어비앤비, 우버, 리프트, 쿠팡등에서 사용 AWS와 구글클라우드와 Azure에서도 지원
  • Airflow에서는 ETL을 DAG라 부르며 웹 인터페이스를 통한 관리 기능 제공
  • 크게 3가지 컴포넌트로 구성됨: 스케줄러, 웹서버, 워커 (Worker)

데이터 웨어하우스의 구성 예

ELT

• ETL: 데이터를 데이터 웨어하우스 외부에서 내부로 가져오는 프로세스
  • 보통 데이터 엔지니어가 이를 수행함
• ELT: 데이터 웨어하우스 내부 데이터를 조작해서 (보통은 좀더 추상화되고 요약된) 새로운 데이터를 만드는 프로세스
  • 이런 프로세스 전용 기술들이 있으며 dbt가 가장 유명: Analytics Engineering
  • 보통 데이터 분석가가 이를 수행함
  • 이 경우 데이터 레이크를 쓰기도 함

데이터 레이크를 포함한 데이터 플랫폼 아키덱처

빅데이터 처리 프레임웍

• 분산 환경 기반 (1대 혹은 그 이상의 서버로 구성)
  • 분산 파일 시스템과 분산 컴퓨팅 시스템이 필요 (2개의 컴포넌트)
• Fault Tolerance
  • 소수의 서버가 고장나도 동작해야함
• 확장이 용이해야함
  • Scale Out이 되어야함
  • 용량을 증대하기 위해서 서버 추가

대표적 빅데이터 프로세싱 시스템

• 1 세대 -> 하둡 기반의 Mapreduce, Hive/Presto
• 2 세대 -> Spark (SQL, DataFrame, Streaming, ML, Graph)

데이터 웨어하우스 옵션들

살펴볼 옵션들

• AWS Redshift
• Snowflake
• Google Cloud BigQuery
• Apache Hive
• Apache Presto
• Apache Iceberg (스토리지이고 Spark이랑 같이쓰면 웨어하우스라고 할 수 있음)
• Apache Spark
• 이 옵션들의 공통점은?
  • Iceberg를 제외하고는 모두 SQL을 지원하는 빅데이터 기반 데이터베이스

AWS Redshift

• 2012년에 시작된 AWS 기반의 데이터웨어하우스로 PB 스케일 데이터 분산 처리 가능
  • Postgresql과 호환되는 SQL로 처리 가능하게 해줌
  • Python UDF (User Defined Function)의 작성을 통해 기능 확장 가능
  • 처음에는 고정비용 모델로 시작했으나 이제는 가변비용 모델도 지원 (Redshift Serverless)
  • 온디맨드 가격 이외에도 예약 가격 옵션도 지원
• CSV, JSON, Avro, Parquet 등과 같은 다양한 데이터 포맷을 지원
• AWS내의 다른 서비스들과 연동이 쉬움
  • S3, DynamoDB, SageMaker 등등 ML 모델의 실행도 지원 (SageMaker)
  • Redshift의 기능 확장을 위해 Redshift Spectrum, AWS Athena등의 서비스와 같이 사용 가능
• 배치 데이터 중심이지만 실시간 데이터 처리 지원
• 웹 콘솔 이외에도 API를 통한 관리/제어 가능

Snowflake

• 2014년에 클라우드 기반 데이터웨어하우스로 시작됨 (2020년 상장)
  • 지금은 데이터 클라우드라고 부를 수 있을 정도로 발전
  • 데이터 판매를 통한 매출을 가능하게 해주는 Data Sharing/Marketplace 제공
  • ETL과 다양한 데이터 통합 기능 제공
• SQL 기반으로 빅데이터 저장, 처리, 분석을 가능하게 해줌
  • 비구조화된 데이터 처리와 머신러닝 기능 제공
• CSV, JSON, Avro, Parquet 등과 같은 다양한 데이터 포맷을 지원
  • S3, GC 클라우드 스토리지, Azure Blog Storage도 지원
• 배치 데이터 중심이지만 실시간 데이터 처리 지원
• 웹 콘솔 이외에도 API를 통한 관리/제어 가능

Apache Hive

• Facebook이 2008년에 시작한 아파치 오픈소스 프로젝트
• 하둡 기반으로 동작하는 SQL 기반 데이터 웨어하우스 서비스
  • HiveQL이라 부르는 SQL 지원
  • MapReduce위에서 동작하는 버전과 Apache Tez를 실행 엔진으로 동작하는 버전 두 가지가 존재
  • 다른 하둡 기반 오픈소스들과 연동이 쉬움 (Spark, HBase 등등)
  • 자바나 파이썬으로 UDF 작성 가능
• CSV, JSON, Avro, Parquet 등과 같은 다양한 데이터 포맷을 지원
• 배치 빅데이터 프로세싱 시스템
  • 데이터 파티셔닝과 버킷팅과 같은 최적화 작업 지원
  • 빠른 처리속도 보다는 처리할 수 있는 데이터 양의 크기에 최적화
• 웹 UI와 커맨드라인 UI (CLI라고 부름) 두 가지를 지원
• 점점 Spark에 의해 밀리는 분위기임

Apache Presto

• Facebook이 2013년에 시작한 아파치 오픈소스 프로젝트
• 다양한 데이터소스에 존재하는 데이터를 대상으로 SQL 실행 가능
  • HDFS (Hadoop Distributed File System), S3, Cassandra, MySQL 등등
  • PrestoSQL이란 부르는 SQL 지원
• CSV, JSON, Avro, ORC, Parquet 등과 같은 다양한 데이터 포맷을 지원
• 배치 빅데이터 프로세싱 시스템
  • Hive와는 다르게 빠른 응답 속도에 좀더 최적화 (메모리 기반)
• 웹 UI와 커맨드라인 UI (CLI라고 부름) 두 가지를 지원
• AWS Athena가 바로 Presto를 기반으로 만들어짐

Apache Iceberg

• Netflix가 2018년에 시작한 아파치 오픈소스 프로젝트로 데이터 웨어하우스 기술이 아님
• 대용량 SCD (Slowly-Changing Datasets) 데이터를 다룰 수 있는 테이블 포맷
  • HDFS, S3, Azure Blob Storage 등의 클라우드 스토리지 지원
  • ACID 트랙잭션과 타임여행 (과거 버전으로 롤백과 변경 기록 유지 등등)
  • 스키마 진화 (Schema Evolution) 지원을 통한 컬럼 제거와 추가 가능 (테이블 재작성 없이)
• 자바와 파이썬 API를 지원
• Spark, Flink, Hive, Hudi 등의 다른 Apache 시스템과 연동 가능

Apache Spark

• UC 버클리 AMPLab이 2013년에 시작한 아파치 오픈소스 프로젝트
• 빅데이터 처리 관련 종합선물세트
  • 배치처리(API/SQL), 실시간처리, 그래프처리, 머신러닝 기능 제공
• 다양한 분산처리 시스템 지원
  • 하둡(YARN), AWS EMR, Google Cloud Dataproc, Mesos, K8s 등등
• 다양한 파일시스템과 연동 가능
  • HDFS, S3, Cassandra, HBase 등등
• CSV, JSON, Avro, ORC, Parquet 등과 같은 다양한 데이터 포맷을 지원
• 다양한 언어 지원: 자바, 파이썬, 스칼라, R

실리콘밸리 회사들의 데이터 스택 트렌드

데이터 플랫폼의 발전단계

• 초기 단계: 데이터 웨어하우스 + ETL
  • 이미 앞에서 살펴봄
• 발전 단계: 데이터 양 증가
  • Spark과 같은 빅데이터 처리시스템 도입
  • 데이터 레이크 도입
• 성숙 단계: 데이터 활용 증대
  • 현업단의 데이터 활용이 가속화
  • ELT 단이 더 중요해지면서 dbt 등의 analytics engineering 도입
  • MLOps 등 머신러닝 관련 효율성 증대 노력 증대

발전 단계: 데이터 양 증가

Spark과 같은 빅데이터 처리시스템 도입

데이터 레이크 도입: 보통 로그 데이터와 같은 대용량 비구조화 데이터 대상

• 데이터 소스 -> 데이터 파이프라인 -> 데이터 웨어하우스
• 데이터 소스 -> 데이터 파이프라인 -> 데이터 레이크
• 데이터 레이크 -> 데이터 파이프라인 -> 데이터 웨어하우스
  • 이때 Spark/Hadoop 등이 사용됨
  • Hadoop: Hive/Presto등이 기반됨

성숙 단계: 현업단의 데이터 활용 가속화

• ELT단이 더 중요해지면서 dbt 등의 analytics engineering 도입
  • 데이터 레이크 to 데이터 레이크, 데이터 레이크 to 데이터 웨어하우스, 데이터 웨어하우스 to 데이터 웨어하우스
• MLOps 등 머신러닝 개발 운영 관련 효율성 증대 노력 증대

실리콘밸리 회사 데이터 스택 비교

• AWS Identity and Access Management(IAM)은 AWS 리소스에 대한 액세스를 안전하게 제어할 수 있는 웹 서비스이다.
• IAM을 사용하여 리소스를 사용하도록 인증(로그인) 및 권한 부여(권한 있음)된 대상을 제어한다.
• AWS 계정을 생성할 때는 해당 계정의 모든 AWS 서비스 및 리소스에 대한 완전한 액세스 권한이 있는 단일 로그인 ID로 시작한다.
• 이 자격 증명은 AWS 계정 루트 사용자라고 하며, 계정을 생성할 때 사용한 이메일 주소와 암호로 로그인하여 액세스한다.
• 일상적인 작업에 루트 사용자를 사용하지 않을 것을 강력히 권장한다.

IAM 특징

• AWS 계정에 대한 공유
• 세분화된 권한
• Amazon EC2에서 실행되는 애플리케이션을 위한 보안 AWS 리소스 액세스
• 멀티 팩터 인증(MFA)
• ID 페더레이션
• 보장을 위한 자격 증명 정보
• PCI DSS 준수
• 많은 AWS 서비스와의 통합
• 최종 일관성
• 무료 사용

IAM 정책

사용자, 그룹, 역할에 대해서 접근 제어를 할 수 있다

S3

• Amazon Simple Storage Service(Amazon S3)는 업계 최고의 확장성, 데이터 가용성, 보안 및 성능을 제공하는 객체 스토리지 서비스.
• 모든 규모와 업종의 고객은 Amazon S3를 사용하여 데이터 레이크, 웹 사이트, 모바일 애플리케이션, 백업 및 복원, 아카이브, 엔터프라이즈 애플리케이션, IoT 디바이스, 빅 데이터 분석 등 다양한 사용 사례에서 원하는 양의 데이터를 저장하고 보호할 수 있다.
• Amazon S3는 특정 비즈니스, 조직 및 규정 준수 요구 사항에 맞게 데이터에 대한 액세스를 최적화, 구조화 및 구성할 수 있는 관리 기능을 제공한다.

  S3 기능

• 스토리지 클래스
• 스토리지 관리
• 액세스 관리
• 데이터 처리
• 스토리지 로깅 및 모니터링
• 분석 및 인사이트
• 강력한 일관성

  Amazon S3를 사용하여 정적 웹 사이트 호스팅

• Amazon S3을 사용하여 정적 웹 사이트를 호스팅할 수 있다.
• 정적 웹 사이트에서 개별 웹 페이지는 정적 콘텐츠를 포함한다.
• 클라이언트 측 스크립트를 포함할 수도 있다.
• 이와는 대조적으로, 동적 웹 사이트는 PHP, JSP 또는 ASP.NET 등 서버 측 스크립트를 포함한 서버 측 처리에 의존한다.

CI / CD

지속적 통합(Continuous Integration) 모든 개발자가 개발한 코드를 공유 리포지토리에 하루에도 여러번 코드를 커밋하고 병합히는 것 지속적 전달(Continuous Delivery) 개발팀이 짧은 주기로 소프트웨어를 개발하고 언제든지 운영환경으로 안정적으로 배포하는 것

CodeCommit

AWS CodeCommit는 클라우드에서 자산 (예: 문서, 소스 코드, 바이너리 파일) 을 비공개로 저장하여 관리하는 데 사용할 수 있도록 Amazon Web Services Services에서 호스팅되는 버전 관리 서비스.

(github과 비슷한 기능)

CodeCommit 특징

• Benefit from a fully managed service hosted by AWS
• Store your code securely
• Work collaboratively on code
• Easily scale your version control projects
• Store anything, anytime
• Integrate with other AWS and third-party services
• Easily migrate files from other remote repositories
• Use the Git tools you already know

CodeBuild

• AWS CodeBuild는 클라우드상의 완전관리형 빌드 서비스.
• CodeBuild는 소스 코드를 컴파일하고 단위 테스트를 실행하며 배포 준비가 완료된 아티팩트를 생성.
• CodeBuild에서는 자체 빌드 서버를 프로비저닝, 관리 및 확장할 필요 없음.
• 이 서비스는 Apache Maven, Gradle 등과 같은 널리 사용되는 프로그래밍 언어 및 빌드 도구에 맞게 사전 패키지된 빌드 환경을 제공.
• CodeBuild에서 빌드 환경을 사용자 지정하여 사용자 고유의 빌드 도구를 사용.
• CodeBuild는 최대 빌드 요청 수에 맞게 자동으로 확장.

CodeBuild 작동방식

CodeDeploy

• CodeDeploy는 Amazon EC2 인스턴스, 온프레미스 인스턴스, 서버리스 Lambda 함수 또는 Amazon ECS 서비스로 애플리케이션 배포를 자동화하는 배포 서비스.

• 다음을 포함하여 다양한 애플리케이션 콘텐츠를 거의 무제한으로 배포가능.

  코드 서버리스 AWS Lambda 함수 웹 및 구성 파일 Executables 패키지 스크립트 멀티미디어 파일

• CodeDeploy는 서버에서 실행되고 Amazon S3 버킷, GitHub 리포지토리 또는 Bitbucket 리포지토리에 저장되는 애플리케이션 콘텐츠를 배포가능. 또한 CodeDeploy는 서버리스 Lambda 함수 배포 가능.

• CodeDeploy를 사용하기 위해 기존 코드를 변경할 필요가 없음.

CodePipeline

AWS CodePipeline은 빠르고 안정적인 애플리케이션 및 인프라 업데이트를 위해 릴리스 파이프라인을 자동화하는 데 도움이 되는 완전관리형의 지속적 전달 서비스.

CodePipeline 특징

• 소프트웨어 릴리스 프로세스를 모델링하고, 서버를 설정하거나 프로비저닝할 필요성을 줄일 수 있음.
• AWS Management Console 또는 AWS command line interface(CLI)를 사용하여 소프트웨어 릴리스 프로세스 단계를 정의할 수 있음.
• 피드백을 반복하고 각 코드 변경을 테스트하여 버그를 포착하는 새로운 기능을 신속하게 릴리스할 수 있음
• 릴리스 프로세스의 모든 단계에서 자체 플러그 또는 사전 구축된 플러그인을 사용하여 필요에 맞추어 조정할 수 있음.

종합 실습

구성도

RDS

• DB 인스턴스는 클라우드에서 실행하는 격리된 데이터베이스 환경
• DB 인스턴스에는 여러 사용자가 만든 데이터베이스가 포함될 수 있으며, 독립 실행형 데이터베이스 인스턴스에 액세스할 때 사용하는 도구 및 애플리케이션을 사용해 액세스할 수 있다.
• AWS 명령줄 도구, Amazon RDS API 작업 또는 AWS Management Console을 사용해 간단히 DB 인스턴스를 만들고 수정할 수 있다.
• 직접 시스템 로그인 불가능.
• RDS 는 serverless 가 아님.

Document DB (NoSQL)

• MongoDB API 워크로드의 완전 관리 및 유연한 확장이 가능한 문서전용(Document) 데이터베이스
• Amazon DocumentDB에서는 스토리지 및 컴퓨팅이 분리되어 각각을 독립적으로 조정.
• 개발자는 데이터 크기에 관계없이 지연 시간이 짧은 읽기 전용 복제본을 몇 분 내에 최대 15개까지 추가하여 읽기 용량을 초당 수백만 개의 요청으로 늘릴 수 있음.
• Amazon DocumentDB는 99.99%의 가용성을 위해 설계되었으며 6개의 데이터 복사본을 3개의 AWS 가용 영역(AZ)에 복제.
• JSON 데이터
• 유연한 인덱싱

MongoDB

• MongoDB는 Document 지향 Database이다.
• 데이터 중복이 발생할 수 있지만, 접근성과 가시성이 좋다
• 스키마 설계가 어렵지만, 스키마가 유연해서 Application의 요구사항에 맞게 데이터를 수용할 수 있다.
• 분산에 대한 솔루션을 자체적으로 지원해서 Scale-out이 쉽다.
• 확장시 ,Application을 변경하지 않아도 된다.

Dynamo DB

• Amazon DynamoDB는 완전관리형 Key-Value 기반 NoSQL 데이터베이스 서비스.
• Auto-Scaling (데이터가 쌓일수록 용량이 증가함)
• DynamoDB는 유휴 시 암호화를 제공하여 중요한 데이터 보호와 관련된 운영 부담 및 복잡성을 제거한다.
• DynamoDB를 통해 원하는 양의 데이터를 저장 및 검색하고 어느 수준의 요청 트래픽도 처리할 수 있는 데이터베이스 테이블을 생성할 수 있다.
• AWS Management Console을 사용하여 리소스 사용률 및 성능 지표를 모니터링할 수 있습니다.
• DynamoDB는 온디맨드 백업 기능을 제공.
• 테이블 생성시 스키마 생성 필요 없음.

Document DB vs Dynamo DB

공통점

• NoSQL Database
• AWS Database Migration Service를 통해 데이터 마이그레이션을 위한 이식성을 제공
• AWS Key Management Service를 통한 저장 데이터 암호화와 보안기능을 제공
• 관리 API 호출과 CloudFormation에 대한 CloudTrail 및 VPC Flow Logs로 감사 기능 제공

Network

Route53

Amazon Route 53는 가용성과 확장성이 뛰어난 DNS(도메인 이름 시스템) 웹 서비스. Route 53을 사용하여 세 가지 주요 기능, 즉 도메인 등록, DNS 라우팅, 상태 확인을 조합하여 실행할 수 있다.

Route53 특징

• Route53 은 public host zone 과 private host zone 존재
• Route53 = DNS(네임서버) + 모니터링 + L4 + GSLB

  Certification Manager

  AWS Certificate Manager(ACM)를 사용하면 AWS 서비스 및 연결된 내부 리소스에 사용할 공인 및 사설 SSL/TLS 인증서를 프로비저닝, 관리 및 배포할 수 있습니다. ACM은 SSL/TLS 인증서를 구매, 업로드 및 갱신하는 데 드는 시간 소모적인 수동 프로세스를 대신 처리해줍니다.

SSL 인증서

SSL 인증서는 공개 키와 개인 키라는 키 쌍을 갖고 있다. 이 키들이 함께 작용하여 암호화된 연결을 수립. 인증서는 또한 "주체(subject)"라는 것을 포함하고 이는 인증서/웹사이트 소유자의 ID이다. 인증서를 얻으려면 서버에서 인증서 서명 요청(CSR)을 생성해야 한다. 이 과정에서 서버에 개인 키와 공개 키 생성. SSL 인증서 발급자(인증 기관 또는 CA라 함)에게 보내는 CSR 데이터 파일에는 공개 키가 포함됩니다.

1. 사용할 TLS/SSL 인증서를 AWS 계정으로 요청하거나 가져온다.
2. 도메인 이름 시스템(DNS) 또는 이메일 검증을 통해 요청된 인증서의 도메인 소유권을 검증하여 인증서 발급을 완료한다.
3. Elastic Load Balancing(ELB), Amazon CloudFront 등과 같은 다양한 AWS 서비스에서 새로 발급되거나 가져온 인증서를 사용한다.

Certification Manager 특징

1. ACM 통합 서비스를 위한 무료 퍼블릭 인증서
2. 관리형 인증서 갱신
3. 손쉽게 인증서 받기

CloudFront

Amazon CloudFront는 뛰어난 성능, 보안 및 개발자 편의를 위해 구축된 콘텐츠 전송 네트워크(CDN) 서비스입니다.

CDN

1. 콘텐츠 전송 네트워크(CDN)는 데이터 사용량이 많은 애플리케이션의 웹 페이지 로드 속도를 높이는 상호 연결된 서버 네트워크
2. 정적 콘텐츠 & 동적 콘텐츠
3. 캐싱 / 동적 가속 / 엣지 로직 계산

CloudFront 특징

1. 대기 시간 감소
2. 보안 향상
3. 비용 절감
4. 사용자 정의 전송

Elastic Load Balancing (ELB)

로드 밸런싱은 애플리케이션을 지원하는 리소스 풀 전체에 네트워크 트래픽을 균등하게 배포하는 방법입니다

Load Balancer

로드밸런서는 서버에 가해지는 부하(=로드)를 분산(=밸런싱)해주는 장치 또는 기술을 통칭한다.

ELB 대상그룹

대상 그룹에 대상을 등록한다. 기본적으로 로드 밸런서는 대상 그룹에 대해 지정한 프로토콜과 포트 번호를 사용하여 등록된 대상으로 요청을 전송한다. 또는 대상 그룹에 각 대상을 등록할 때 이 포트를 재정의할 수 있다.

VPC

Amazon Virtual Private Cloud(Amazon VPC)를 이용하면 사용자가 정의한 가상 네트워크로 AWS 리소스를 시작할 수 있다. 이 가상 네트워크는 AWS의 확장 가능한 인프라를 사용한다는 이점과 함께 고객의 자체 데이터 센터에서 운영하는 기존 네트워크와 유사하다.

VPC 기능

Virtual Private Cloud(VPC) : VPC는 자체 데이터 센터에서 운영하는 기존 네트워크와 아주 유사한 가상 네트워크이다. VPC를 생성한 후 서브넷을 추가할 수 있다.

서브넷: 서브넷은 VPC의 IP 주소 범위이다. 서브넷은 단일 가용 영역에 상주해야 합니다. 서브넷을 추가한 후에는 VPC에 AWS 리소스 배포할 수 있다.

IP 주소 지정: VPC와 서브넷에 IPv4 주소와 IPv6 주소를 할당할 수 있다. 또한 퍼블릭 IPv4 및 IPv6 GUA 주소를 AWS로 가져오고 VPC의 리소스(예: EC2 인스턴스, NAT 게이트웨이, Network Load Balancer)에 할당할 수 있다.

라우팅: 라우팅 테이블을 사용하여 서브넷 또는 게이트웨이의 네트워크 트래픽이 전달되는 위치를 결정한다.

게이트웨이 및 엔드포인트: 게이트웨이는 VPC를 다른 네트워크에 연결한다. 예를 들면, 인터넷 게이트웨이를 사용하여 VPC를 인터넷에 연결한다. VPC 엔드포인트를 사용하여 인터넷 게이트웨이 또는 NAT 장치를 사용하지 않고 AWS 서비스에 비공개로 연결합니다.

피어링 연결: VPC 피어링 연결을 사용하여 두 VPC의 리소스 간 트래픽을 라우팅한다.

트래픽 미러링: 네트워크 인터페이스에서 네트워크 트래픽을 복사하고 심층 패킷 검사를 위해 보안 및 모니터링 어플라이언스로 전송한다.

Transit Gateway: 중앙 허브 역할을 하는 전송 게이트웨이를 사용하여 VPC, VPN 연결 및 AWS Direct Connect 연결 간에 트래픽을 라우팅한다.

VPC 흐름 로그: 흐름 로그는 VPC의 네트워크 인터페이스로 들어오고 나가는 IP 트래픽에 대한 정보를 캡처한다.

VPN 연결: AWS Virtual Private Network(AWS VPN)을 사용하여 온프레미스 네트워크에 VPC를 연결한다.

CIDR

CIDR(Classless Inter-Domain Routing, 사이더) : 클래스 없는 도메인 간 라우팅 기법 네트워크 영역과 호스팅 영역을 나눔 (어느 대역까지 IP를 쓸 수 있는가를 표현)

VPC 생성

Private IP 대역

10.0.0.010.255.255.255 172.16.0.0172.31.255.255 192.168.0.0~192.168.255.255

Amazon Web Services(AWS)

• 전 세계적으로 분포한 데이터 센터에서 200개가 넘는 완벽한 기능의 서비스를 제공하는, 세계적으로 가장 포괄적이며, 널리 채택되고 있는 클라우드 플랫폼.
• 빠르게 성장하는 스타트업, 가장 큰 규모의 엔터프라이즈, 주요 정부 기관을 포함하여 수백만명의 고객이 AWS를 사용하여 비용을 절감하고, 민첩성을 향상시키고 더 빠르게 혁신함.

  클라우드 컴퓨팅이란?

• 클라우드 컴퓨팅 IT 리소스를 인터넷을 통해 온디맨드로 제공하고 사용한 만큼만 비용을 지불하는 방식.
• 물리적 데이터 센터와 서버를 구입, 소유 및 유지 관리하는 대신, Amazon Web Services(AWS)와 같은 클라우드 공급자로부터 필요에 따라컴퓨팅 파워, 스토리지, 데이터베이스와 같은 기술 서비스에 액세스 한다.

  클라우드 컴퓨팅 이점

  • 민첩성

• 탄력성
• 비용절감
• On demand
• 관리 용이성

클라우드 유형

클라우드 서비스 제품

• 아마존 AWS (Amazon Web Service) aws.amazon.com/ • 마이크로소프트 애저 (Azure) azure.microsoft.com/ • 구글 GCP (Google Cloud Platform) cloud.google.com/ • 오라클 OCI (Oracle Cloud Infrastructure) www.oracle.com/cloud/ • IBM 클라우드 (IBM Cloud) www.ibm.com/kr-ko/cloud • 알리바바 클라우드 (Alibaba Cloud) www.alibabacloud.com/ • KT 클라우드 (KT Cloud) cloud.kt.com/ • 네이버 NCP (Naver Cloud Platform) www.ncloud.com/

AWS 기본용어

가상화 : 물리적 컴퓨터 하드웨어를 보다 효율적으로 활용할 수 있도록 해주는 프로세스이며, 이는 클라우드 컴퓨팅의 기반을 제공하는 기술 가상머신 : 가상 머신(VM)은 소프트웨어 형식으로 물리적 컴퓨팅을 시뮬레이션하는 가상 환경이다. 이들은 일반적으로 VM의 구성, 가상 하드 드라이브의 스토리지, 그리고 특정 시점에 해당 상태를 유지하는 VM의 일부 스냅샷을 포함한 다수의 파일들로 구성되어 있다. 스냅샷 :

• 스냅샷은 마치 사진 찍듯이 특정 시점에 스토리지의 파일 시스템을 포착해 보관하는 기술
• Windows OS의 복원 지점과 같이 장애나 데이터 손상 시 스냅샷을 생성한 시점으로 데이터를 복구
• 스냅샷은 원본 데이터를 그대로 복사해 다른 곳에 저장하는 백업과 달리 초기 생성 시 혹은 데이터의 변경이 있기 전까지는 스토리지의 공간을 차지하지 않는다.
• 메타데이터(데이터에 대한 부가적인 정보)의 복사본에 해당하기 때문에 생성하는 데 오랜 시간이 걸리지 않고, 장애 상황이 발생해도 빠르게 데이터를 복원 데이터 센터 : 수많은 서버들을 한데 모아 네트워크로 연결해 놓은 시설

Region(지역) :

• Region 은 Data Center가 위치한 지역
• IT 리소스를 생성할 Region은 선택 가능
• 대상 고객의 지역과 자원 생성할 Region이 최대한 가까워야 함
• 국가마다 자원사용 비용이 다름

Availability Zone(가용영역) :

• 하나의 Region 은 두 개 이상의 Availability Zone 으로 구성됨.
• 줄여서 AZ 로 표시

EC2

AWS EC2 기능

인스턴스: 가상 컴퓨팅 환경 Amazon 머신 이미지(AMI): 서버에 필요한 운영체제와 여러 소프트웨어들이 적절히 구성된 상태로 제공되는 템플릿으로 인스턴스를 쉽게 만들 수 있습니다. 인스턴스 유형: 인스턴스를 위한 CPU, 메모리, 스토리지, 네트워킹 용량의 여러 가지 구성 제공 키 페어를 사용하여 인스턴스 로그인 정보 보호(AWS는 퍼블릭 키를 저장하고 사용자는 개인 키를 안전한 장소에 보관하는 방식) 인스턴스 스토어 볼륨: 임시 데이터를 저장하는 스토리지 볼륨으로 인스턴스 중단, 최대 절전 모드로 전환 또는 종료 시 삭제됨 Amazon Elastic Block Store(Amazon EBS), 즉 Amazon EBS 볼륨을 사용해 영구 스토리지 볼륨에 데이터 저장 보안 그룹을 사용해 인스턴스에 연결할 수 있는 프로토콜, 포트, 소스 IP 범위를 지정하는 방화벽 기능 탄력적 IP 주소(EIP): 동적 클라우드 컴퓨팅을 위한 고정 IPv4 주소 태그: 사용자가 생성하여 Amazon EC2 리소스에 할당할 수 있는 메타데이터 Virtual Private Clouds(VPC) : AWS 클라우드에서는 논리적으로 격리되어 있지만 원할 때마다 고객의 네트워크와 간편히 연결할 수 있는 가상 네트워크

Elasticbeanstalk

• Elastic Beanstalk를 사용하면 애플리케이션을 실행하는 인프라에 대해 자세히 알지 못해도 AWS 클라우드에서 애플리케이션을 신속하게 배포하고 관리할 수 있습니다.
• Elastic Beanstalk를 사용하면 선택 또는 제어에 대한 제한 없이 관리 복잡성을 줄일 수 있습니다.
• 애플리케이션을 업로드하기만 하면 Elastic Beanstalk에서 용량 프로비저닝, 로드 밸런싱, 조정, 애플리케이션 상태 모니터링에 대한 세부 정보를 자동으로 처리합니다.
• Elastic Beanstalk는 Go, Java, .NET, Node.js, PHP, Python 및 Ruby에서 개발된 애플리케이션을 지원합니다.
• 애플리케이션을 배포할 때, Elastic Beanstalk가 선택된 지원 가능 플랫폼 버전을 구축하고 Amazon EC2 등의 AWS 리소스를 하나 이상 프로비저닝하여 애플리케이션을 실행합니다.

(1) CTE를 빌딩블록으로 사용

WITH first AS (
 SELECT userid, ts, channel, ROW_NUMBER() OVER(PARTITION BY userid ORDER BY ts) seq
 FROM raw_data.user_session_channel usc
 JOIN raw_data.session_timestamp st ON usc.sessionid = st.sessionid
), last AS (
 SELECT userid, ts, channel, ROW_NUMBER() OVER(PARTITION BY userid ORDER BY ts DESC) seq
 FROM raw_data.user_session_channel usc
 JOIN raw_data.session_timestamp st ON usc.sessionid = st.sessionid
) 
SELECT first.userid AS userid, first.channel AS first_channel, last.channel AS last_channel
FROM first
JOIN last ON first.userid = last.userid and last.seq = 1
WHERE first.seq = 1;

(2) JOIN 방식

SELECT first.userid AS userid, first.channel AS first_channel, last.channel AS last_channel
FROM (
 SELECT userid, ts, channel, ROW_NUMBER() OVER(PARTITION BY userid ORDER BY ts) seq
 FROM raw_data.user_session_channel usc
 JOIN raw_data.session_timestamp st ON usc.sessionid = st.sessionid
) first
JOIN (
 SELECT userid, ts, channel, ROW_NUMBER() OVER(PARTITION BY userid ORDER BY ts DESC) seq
 FROM raw_data.user_session_channel usc
 JOIN raw_data.session_timestamp st ON usc.sessionid = st.sessionid
) last ON first.userid = last.userid and last.seq = 1
WHERE first.seq = 1;

(3) GROUP BY 방식

SELECT userid,
 MAX(CASE WHEN rn1 = 1 THEN channel END) first_touch,
 MAX(CASE WHEN rn2 = 1 THEN channel END) last_touch
FROM (
 SELECT userid,
 channel,
 (ROW_NUMBER() OVER (PARTITION BY usc.userid ORDER BY st.ts asc)) AS rn1,
 (ROW_NUMBER() OVER (PARTITION BY usc.userid ORDER BY st.ts desc)) AS rn2
 FROM raw_data.user_session_channel usc
 JOIN raw_data.session_timestamp st ON usc.sessionid = st.sessionid
)
GROUP BY 1;

(4) FIRST_VALUE/LAST_VALUE

--userid 마다 결과가 중복이 되기 때문에 GROUP BY 보다 퍼포먼스가 더 떨어짐
SELECT DISTINCT
 A.userid,
 FIRST_VALUE(A.channel) over(partition by A.userid order by B.ts
rows between unbounded preceding and unbounded following) AS First_Channel,
 LAST_VALUE(A.channel) over(partition by A.userid order by B.ts
rows between unbounded preceding and unbounded following) AS Last_Channel
FROM raw_data.user_session_channel A
LEFT JOIN raw_data.session_timestamp B ON A.sessionid = B.sessionid;

Gross Revenue가 가장 큰 UserID 10개 찾기

• user_session_channel과 session_transaction과 session_timestamp 테이블을 사용

• Gross revenue: Refund 포함한 매출

  (1) GROUP BY

  SELECT
  userID,
  SUM(amount)
  FROM raw_data.session_transaction st
  LEFT JOIN raw_data.user_session_channel usc ON st.sessionid = usc.sessionid
  GROUP BY 1
  ORDER BY 2 DESC
  LIMIT 10;

  (2) SUM OVER

  ```sql SELECT DISTINCT usc.userid, SUM(amount) OVER(PARTITION BY usc.userid) FROM raw_data.user_session_channel AS usc JOIN raw_data.session_transaction AS revenue ON revenue.sessionid = usc.sessionid ORDER BY 2 DESC LIMIT 10;

• • (1)번 방식이 더 깔끔, 퍼포먼스가 더 좋음 중복이 없기 때문 ```

    raw_data.nps 테이블을 바탕으로 월별 NPS 계산

• 고객들이 0 (의향 없음) 에서 10 (의향 아주 높음)

  • detractor (비추천자) : 0 에서 6
  • passive (소극자) : 7이나 8점
  • promoter (홍보자) : 9나 10점

• NPS = promoter 퍼센트 - detractor 퍼센트

• 10점 만점으로 '주변에 추천하겠는가?'라는 질문을 기반으로 고객 만족도를 계산

  • 10, 9점 추천하겠다는 고객(promoter)의 비율에서 0-6점의 불평고객(detractor)의 비율을 뺀 것이 NPS

    (1)

    SELECT month, 
    ROUND((promoters-detractors)::float/total_count*100, 2) AS overall_nps
    FROM (
    SELECT LEFT(created, 7) AS month,
    COUNT(CASE WHEN score >= 9 THEN 1 END) AS promoters,
    COUNT(CASE WHEN score <= 6 THEN 1 END) AS detractors,
    COUNT(CASE WHEN score > 6 AND score < 9 THEN 1 END) As passives,
    COUNT(1) AS total_count
    FROM raw_data.nps
    GROUP BY 1
    ORDER BY 1
    );

    (2)

    SELECT LEFT(created, 7) AS month,
    ROUND(SUM(CASE
    WHEN score >= 9 THEN 1 
    WHEN score <= 6 THEN -1 END)::float*100/COUNT(1), 2)
    FROM raw_data.nps
    GROUP BY 1
    ORDER BY 1;

    트랜잭션 소개와 실습

    트랜잭션이란?

• Atomic하게 실행되어야 하는 SQL들을 묶어서 하나의 작업처럼 처리하는 방법

  • 이는 DDL이나 DML 중 레코드를 수정/추가/삭제한 것에만 의미가 있음.
  • SELECT에는 트랜잭션을 사용할 이유가 없음
  • BEGIN과 END 혹은 BEGIN과 COMMIT 사이에 해당 SQL들을 사용
  • ROLLBACK

• 은행 계좌 이체가 아주 좋은 예

  • 계좌 이체: 인출과 입금의 두 과정으로 이뤄짐
  • 만일 인출은 성공했는데 입금이 실패한다면?
  • 이 두 과정은 동시에 성공하던지 실패해야함 -> Atomic하다는 의미
  • 이런 과정들을 트랜잭션으로 묶어주어야함
  • 조회만 한다면 이는 트랜잭션으로 묶일 이유가 없음

    BEGIN; A의 계좌로부터 인출; B의 계좌로 입금; END;

• END와 COMMIT은 동일

• 만일 BEGIN 전의 상태로 돌아가고 싶다면 ROLLBACK 실행

• 이 동작은 commit mode에 따라 달라짐!

트랜잭션 커밋 모드: autocommit

• autocommit = True
  • 모든 레코드 수정/삭제/추가 작업이 기본적으로 바로 데이터베이스에 쓰여짐. 이를 커밋(Commit)된다고 함.
  • 만일 특정 작업을 트랜잭션으로 묶고 싶다면 BEGIN과 END(COMMIT)/ROLLBACK으로 처리
• autocommit = False
  • 모든 레코드 수정/삭제/추가 작업이 COMMIT 호출될 때까지 커밋되지 않음

    트랜잭션 방식

• Google Colab의 트랜잭션
  • 기본적으로 모든 SQL statement가 바로 커밋됨 (autocommit=True)
  • 이를 바꾸고 싶다면 BEGIN;END; 혹은 BEGIN;COMMIT을 사용 (혹은 ROLLBACK;)
• psycopg2의 트랜잭션
  • autocommit이라는 파라미터로 조절가능
  • autocommit=True가 되면 기본적으로 PostgreSQL의 커밋 모드와 동일
  • autocommit=False가 되면 커넥션 객체의 .commit()과 .rollback()함수로 트랜잭션 조절 가능
  • 무엇을 사용할지는 개인 취향

    DELETE FROM vs. TRUNCATE

    DELETE FROM table_name (not DELETE * FROM)

• 테이블에서 모든 레코드를 삭제
• vs. DROP TABLE table_name
• WHERE 사용해 특정 레코드만 삭제 가능:
  • DELETE FROM raw_data.user_session_channel WHERE channel = ‘Google’

    TRUNCATE table_name도 테이블에서 모든 레코드를 삭제

• DELETE FROM은 속도가 느림
• TRUNCATE이 전체 테이블의 내용 삭제시에는 여러모로 유리
• 하지만 두가지 단점이 존재
  • TRUNCATE는 WHERE을 지원하지 않음
  • TRUNCATE는 Transaction을 지원하지 않음 (Rollback 이 안됨)

    알아두면 유용한 SQL 문법들

• UNION, EXCEPT, INTERSECT
• COALESCE, NULLIF
• LISTAGG
• LAG
• WINDOW 함수
  • ROW_NUMBER OVER
  • SUM OVER
  • FIRST_VALUE, LAST_VALUE
• JSON Parsing 함수

  UNION, EXCEPT, INTERSECT

  UNION (합집합)

• 여러개의 테이블들이나 SELECT 결과를 하나의 결과로 합쳐줌
• UNION vs. UNION ALL
  • UNION은 중복을 제거

    EXCEPT (MINUS)

• 하나의 SELECT 결과에서 다른 SELECT 결과를 빼주는 것이 가능

  INTERSECT (교집합)

• 여러 개의 SELECT문에서 같은 레코드들만 찾아줌

COALESCE, NULLIF

COALESCE(Expression1, Expression2, …):

• 첫번째 Expression부터 값이 NULL이 아닌 것이 나오면 그 값을 리턴하고 모두 NULL이면 NULL을 리턴한다.
• NULL값을 다른 값으로 바꾸고 싶을 때 사용한다.

  NULLIF(Expression1, Expression2):

• Expression1과 Expression2의 값이 같으면 NULL을 리턴한다

  LISTAGG

• GROUP BY에서 사용되는 Aggregate 함수 중의 하나
• 사용자 ID별로 채널을 순서대로 리스트:

  SELECT
  userid,
  LISTAGG(channel) WITHIN GROUP (ORDER BY ts) channels
  FROM raw_data.user_session_channel usc
  JOIN raw_data.session_timestamp st ON usc.sessionid = st.sessionid
  GROUP BY 1
  LIMIT 10;

  SELECT
  userid,
  LISTAGG(channel, '->') WITHIN GROUP (ORDER BY ts) channels
  FROM raw_data.user_session_channel usc
  JOIN raw_data.session_timestamp st ON usc.sessionid = st.sessionid
  GROUP BY 1
  LIMIT 10;

  WINDOW

• Syntax:

  function(expression) OVER ( [ PARTITION BY expression] [ ORDER BY expression ])

• Useful functions:
  • ROW_NUMBER, FIRST_VALUE, LAST_VALUE, LAG
  • Math functions: AVG, SUM, COUNT, MAX, MIN, MEDIAN, NTH_VALUE

    LAG 함수

• 어떤 사용자 세션에서 시간순으로 봤을 때
  • 앞 세션의 채널이 무엇인지 알고 싶다면?
  • 혹은 다음 세션의 채널이 무엇인지 알고 싶다면? ```sql
• • 이전 채널 찾기 SELECT usc.*, st.ts, LAG(channel,1) OVER (PARTITION BY userId ORDER BY ts) prev_channel FROM raw_data.user_session_channel usc JOIN raw_data.session_timestamp st ON usc.sessionid = st.sessionid ORDER BY usc.userid, st.ts ```
• • 다음 채널을 찾으려면?

    JSON Parsing Functions

• JSON의 포맷을 이미 아는 상황에서만 사용가능한 함수
  • JSON String을 입력으로 받아 특정 필드의 값을 추출가능 (nested 구조 지원)
• 예제) JSON_EXTRACT_PATH_TEXT
  • SELECT JSON_EXTRACT_PATH_TEXT('{"f2":{"f3":"1"},"f4":{"f5":"99","f6":"star"}}','f4', 'f6');

    {
    "f2":{
    "f3":"1"
    },
    "f4":{
    "f5":"99",
    "f6":"star"
    }
    }

SQL 조인은 두 개 혹은 그 이상의 테이블들을 공통 필드를 가지고 머지하는데 사용된다. 이는 스타 스키마로 구성된 테이블들로 분산되어 있던 정보를 통합하는데 사용된다.

왼쪽 테이블을 LEFT라고 하고 오른쪽 테이블을 RIGHT이라고 하자. JOIN의 결과는 방식에 상관없이 양쪽의 필드를 모두 가진 새로운 테이블을 만들어내게 된다. 조인의 방식에 따라 다음 두 가지가 달라진다:

1. 어떤 레코드들이 선택되는지?

2. 어떤 필드들이 채워지는지?

다양한 종류의 조인

Source: https://theartofpostgresql.com/blog/2019-09-sql-joins/

JOIN 문법

SELECT A.*, B.*
FROM raw_data.table1 A
____ JOIN raw_data.table2 B ON A.key1 = B.key1 and A.key2 = B.key2
--INNER, FULL, LEFT, RIGHT, CROSS (대부분 INNER와 LEFT 사용)
WHERE A.ts >= '2019-01-01';

JOIN시 고려해야할 점

먼저 중복 레코드가 없고 Primary Key의 uniqueness가 보장됨을 체크

• 아주 중요함!!!

조인하는 테이블들간의 관계를 명확하게 정의

One to one

• 완전한 one to one: user_session_channel & session_timestamp
• 한쪽이 부분집합이 되는 one to one: user_session_channel & session_transaction

  One to many? (order vs order_items)

• 이 경우 중복이 더 큰 문제됨 -> 증폭!!

  Many to one?

• 방향만 바꾸면 One to many로 보는 것과 사실상 동일.

  Many to many?

• 이런 경우는 많지 않으며 이는 one to one이나 one to many로 바꾸는 것이 가능하다면 변환하여 조인하는 것이 덜 위험 어느 테이블을 베이스로 잡을지 (From에 사용할지) 결정해야함

예제 테이블

INNER JOIN

1. 양쪽 테이블에서 매치가 되는 레코드들만 리턴함
2. 양쪽 테이블의 필드가 모두 채워진 상태로 리턴됨

   SELECT * FROM raw_data.Vital v
   JOIN raw_data.Alert a ON v.vitalID = a.vitalID;

   LEFT JOIN

3. 왼쪽 테이블(Base)의 모든 레코드들을 리턴함
4. 오른쪽 테이블의 필드는 왼쪽 레코드와 매칭되는 경우에만 채워진 상태로 리턴됨

   SELECT * FROM raw_data.Vital v
   LEFT JOIN raw_data.Alert a ON v.vitalID = a.vitalID;

   FULL JOIN

5. 왼쪽 테이블과 오른쪽 테이블의 모든 레코드들을 리턴함
6. 매칭되는 경우에만 양쪽 테이블들의 모든 필드들이 채워진 상태로 리턴됨

   SELECT * FROM raw_data.Vital v
   FULL JOIN raw_data.Alert a ON v.vitalID = a.vitalID;

   CROSS JOIN

7. 왼쪽 테이블과 오른쪽 테이블의 모든 레코드들의 조합을 리턴함

   SELECT * FROM raw_data.Vital v CROSS JOIN raw_data.Alert a;

   SELF JOIN

8. 동일한 테이블을 alias를 달리해서 자기 자신과 조인함

   SELECT * FROM raw_data.Vital v1
   JOIN raw_data.Vital v2 ON v1.vitalID = v2.vitalID;

BOOLEAN 타입 처리

• True or False
• 다음 2개는 동일한 표현
  • flag = True
  • flag is True
• 다음 2개는 동일한 표현인가? (수학적으로는 같지만 SQL에서는 같지 않다)
  • flag is True
  • flag is not False

    SELECT
    COUNT(CASE WHEN flag = True THEN 1 END) true_cnt1,
    COUNT(CASE WHEN flag is True THEN 1 END) true_cnt2,
    COUNT(CASE WHEN flag is not False THEN 1 END) not_false_cnt
    FROM raw_data.boolean_test;

    NULL 비교

• NULL 비교는 항상 IS 혹은 IS NOT으로 수행
• NULL 비교를 = 혹은 != 혹은 <>으로 수행하면 잘못된 결과가 나옴

  SELECT COUNT(1)
  FROM raw_data.boolean_test
  WHERE flag is NULL;
  SELECT COUNT(1)
  FROM raw_data.boolean_test
  WHERE flag = NULL;

  채널별 월 매출액 테이블 만들기 (숙제)

1. 먼저 유일한 사용자 수부터 세보자

   SELECT LEFT(ts, 7) "month",
   usc.channel,
   COUNT(DISTINCT userid) uniqueUsers
   FROM raw_data.user_session_channel usc
   JOIN raw_data.session_timestamp t ON t.sessionid = usc.sessionid
   GROUP BY 1, 2
   ORDER BY 1, 2;

   복잡한 JOIN시 먼저 JOIN 전략부터 수립

   raw_data.user_session_channel raw_data.session_timestamp raw_data.session_transaction

• 의 3개 테이블 모두 sessionid를 기반으로 조인을 해야함
• user_session_channel과 session_timestamp는 일대일로 조인가능: INNER JOIN
• 하지만 session_transaction의 경우에는 모든 sessionid가 존재하지 않음
  • LEFT JOIN (혹은 RIGHT JOIN)
  • FROM에 사용하는 테이블은 user_session_channel 혹은 session_timestamp가 되어야함

2. 이제 session_transaction 테이블을 추가해보자

   SELECT LEFT(ts, 7) "month",
   usc.channel,
   COUNT(DISTINCT userid) uniqueUsers
   FROM raw_data.user_session_channel usc
   JOIN raw_data.session_timestamp t ON t.sessionid = usc.sessionid
   LEFT JOIN raw_data.session_transaction st ON st.sessionid = usc.sessionid
   GROUP BY 1, 2
   ORDER BY 1, 2;

3. 이제 paidUsers를 추가해보자

   SELECT LEFT(ts, 7) "month",
   usc.channel,
   COUNT(DISTINCT userid) uniqueUsers,
   COUNT(DISTINCT CASE WHEN amount > 0 THEN usc.userid END) paidUsers,
   FROM raw_data.user_session_channel usc
   JOIN raw_data.session_timestamp t ON t.sessionid = usc.sessionid
   LEFT JOIN raw_data.session_transaction st ON st.sessionid = usc.sessionid
   GROUP BY 1, 2
   ORDER BY 1, 2;

4. 이제 conversionRate을 추가해보자

   첫 번째 시도:

• paidUsers/uniqueUsers AS conversionRate

  두 번째 시도:

• paidUsers::float/uniqueUsers AS conversionRate

  세 번째 시도: (소숫점 둘째까지)

• ROUND(paidUsers*100.0/uniqueUsers, 2) AS conversionRate

  네 번째 시도: (NULLIF)

• ROUND(paidUsers*100.0/NULLIF(uniqueUsers, 0), 2) AS conversionRate

NULLIF

paidUsers/uniqueUsers

• 0으로 나누는 경우 divide by 0 에러 발생
• 이를 어떻게 방지할까? NULLIF를 사용하여 0을 NULL로 변경
  • paidUsers/NULLIF(uniqueUsers, 0)
  • 다시 한번 사칙연산에 NULL이 들어가면 결과도 NULL이 됨을 기억!

    SELECT LEFT(ts, 7) "month", -- "year month"
    channel,
    COUNT(DISTINCT usc.userid) uniqueUsers,
    COUNT(DISTINCT CASE WHEN amount > 0 THEN usc.userid END) paidUsers,
    ROUND(paidUsers::float*100/NULLIF(uniqueUsers, 0),2) conversionRate,
    SUM(amount) grossRevenue,
    SUM(CASE WHEN refunded is False THEN amount END) netRevenue
    FROM raw_data.user_session_channel usc
    LEFT JOIN raw_data.session_timestamp t ON t.sessionid = usc.sessionid
    LEFT JOIN raw_data.session_transaction st ON st.sessionid = usc.sessionid
    GROUP BY 1, 2
    ORDER BY 1, 2;

    COALESCE

• NULL 값을 다른 값으로 바꿔주는 함수
  • 즉 NULL대신에 다른 백업값을 리턴해주는 함수
• COALESCE(exp1, exp2, exp3, …)
  • exp1부터 인자를 하나씩 살펴서 NULL이 아닌 값이 나오면 그걸 리턴
  • 끝까지 갔는데도 모두 NULL이면 최종적으로 NULL을 리턴

    SELECT
    value,
    COALESCE(value, 0) -- value가 NULL이면 0을 리턴
    FROM raw_data.count_test;

    공백 혹은 예약키워드를 필드 이름으로 사용하려면?

• ""로 둘러싸서 사용

  CREATE TABLE keeyong.test (
  group int primary key,
  'mailing address' varchar(32)
  );

  결과물

  DROP TABLE IF EXISTS adhoc.keeyong_monthly_channel_summary;
  CREATE TABLE adhoc.keeyong_monthly_channel_summary AS
  SELECT LEFT(ts, 7) "month", 
  channel,
  COUNT(DISTINCT usc.userid) uniqueUsers,
  COUNT(DISTINCT CASE WHEN amount > 0 THEN usc.userid END) paidUsers,
  ROUND(paidUsers::float*100/NULLIF(uniqueUsers, 0),2) conversionRate,
  SUM(amount) grossRevenue,
  SUM(CASE WHEN refunded is False THEN amount END) netRevenue
  FROM raw_data.user_session_channel usc
  LEFT JOIN raw_data.session_timestamp t ON t.sessionid = usc.sessionid
  LEFT JOIN raw_data.session_transaction st ON st.sessionid = usc.sessionid
  GROUP BY 1, 2;

테이블의 레코드를 그룹핑하여 그룹별로 다양한 정보를 계산

이는 두 단계로 이뤄짐

• 먼저 그룹핑을 할 필드를 결정 (하나 이상의 필드가 될 수 있음)
  • GROUP BY로 지정 (필드 이름을 사용하거나 필드 일련번호를 사용)
• 다음 그룹별로 계산할 내용을 결정
  • 여기서 Aggregate함수를 사용
  • COUNT, SUM, AVG, MIN, MAX, LISTAGG, …
  • 보통 필드 이름을 지정하는 것이 일반적 (alias)

    월별 세션수를 계산하는 SQL

• raw_data.session_timestamp를 사용 (sessionId와 ts 필드)

  SELECT
  LEFT(ts, 7) AS mon, -- left를 쓰는 순간 ts가 문자열로 바뀜 7개 빼면 yyyy-mm 가져옴
  COUNT(1) AS session_count
  FROM raw_data.session_timestamp
  GROUP BY 1 -- GROUP BY mon, GROUP BY LEFT(ts, 7)
  ORDER BY 1;

-> 앞서 설명한 raw_data.session_timestamp raw_data.user_session_channel 테이블들을 사용해서 다음을 계산하는 SQL을 만들어 보자

가장 많이 사용된 채널은 무엇인가?

• 가장 많이 사용되었다는 정의는?
  • 사용자 기반 아니면 세션 기반?
• 필요한 정보 - 채널 정보, 사용자 정보 혹은 세션 정보
• 먼저 어느 테이블을 사용해야하는지 생각!
  • user_session_channel?
  • session_timestamp?
  • 혹은 이 2개의 테이블을 조인해야하나?

SELECT
 channel,
 COUNT(1) AS session_count,
 COUNT(DISTINCT userId) AS user_count
FROM raw_data.user_session_channel
GROUP BY 1 -- GROUP BY channel 숫자를 쓰는 것을 선호
ORDER BY 2 DESC; -- ORDER BY session_count DESC

가장 많은 세션을 만들어낸 사용자 ID는 무엇인가?

• 필요한 정보 - 세션 정보, 사용자 정보
• 먼저 어느 테이블을 사용해야하는지 생각!
  • user_session_channel?
  • session_timestamp?
  • 혹은 이 2개의 테이블을 조인해야하나?

    SELECT
    userId,
    COUNT(1) AS count
    FROM raw_data.user_session_channel
    GROUP BY 1 -- GROUP BY userId
    ORDER BY 2 DESC -- ORDER BY count DESC
    LIMIT 1;

    월별 유니크한 사용자 수

• 이게 바로 MAU(Monthly Active User)에 해당
• 필요한 정보 - 시간 정보, 사용자 정보
• 먼저 어느 테이블을 사용해야하는지 생각!
  • user_session_channel (userId, sessionId, channel)?
  • session_timestamp (sessionId, ts)?
  • 혹은 이 2개의 테이블을 조인해야하나? ```sql
• -내부 조인(inner join) SELECT TO_CHAR(A.ts, 'YYYY-MM') AS month, COUNT(DISTINCT B.userid) AS mau FROM raw_data.session_timestamp A JOIN raw_data.user_session_channel B ON A.sessionid = B.sessionid GROUP BY 1 ORDER BY 1 DESC; ```

  TO_CHAR (A.ts, ‘YYYY-MM’)와 같은 기능을 하는 코드

• LEFT(A.ts, 7)
• DATE_TRUNC(‘month’, A.ts) : ts타입의 값을 리턴
• SUBSTRING(A.ts, 1, 7)

  필드/테이블 이름에 Alias 사용. AS는 필수가 아님.

• COUNT(DISTINCT B.userid) AS mau와
• COUNT(DISTINCT B.userid) mau는 동일

  ORDER BY와 GROUP BY:

• 포지션 번호 vs. 필드 이름
  • GROUP BY 1 == GROUP BY month == GROUP BY TO_CHAR(A.ts, 'YYYY-MM'0)

월별 채널별 유니크한 사용자 수

• 필요한 정보 - 시간 정보, 사용자 정보, 채널 정보
• 먼저 어느 테이블을 사용해야하는지 생각!
  • user_session_channel (userId, sessionId, channel)?
  • session_timestamp (sessionId, ts)?
  • 혹은 이 2개의 테이블을 조인해야하나?

    SELECT 
    TO_CHAR(A.ts, 'YYYY-MM') AS month,
    channel,
    COUNT(DISTINCT B.userid) AS mau
    FROM raw_data.session_timestamp A
    JOIN raw_data.user_session_channel B ON A.sessionid = B.sessionid
    GROUP BY 1, 2 --월별 코드보다 그룹이 하나 늘었다
    ORDER BY 1 DESC, 2;

    CTAS와 CTE 소개

    CTAS: SELECT를 가지고 테이블 생성

• 간단하게 새로운 테이블을 만드는 방법
• 자주 조인하는 테이블들이 있다면 이를 CTAS를 사용해서 조인해두면 편리해짐 ```sql
• -충돌을 막기 위해 summary 앞에 영문이름 넣음 DROP TABLE IF EXISTS adhoc.keeyong_session_summary; CREATE TABLE adhoc.keeyong_session_summary AS SELECT B.*, A.ts FROM raw_data.session_timestamp A JOIN raw_data.user_session_channel B ON A.sessionid = B.sessionid;

  ### 월별 유니크한 사용자 수를 다시 풀어보기
  ```sql
  SELECT 
  TO_CHAR(ts, 'YYYY-MM') AS month,
  COUNT(DISTINCT userid) AS mau
  FROM adhoc.keeyong_session_summary
  GROUP BY 1 
  ORDER BY 1 DESC; 

  항상 시도해봐야하는 데이터 품질 확인 방법들

  1. 중복된 레코드들 체크하기

• 다음 두 개의 카운트를 비교 ```sql SELECT COUNT(1) FROM adhoc.keeyong_session_summary;
• • 문제가 없으면 동일해야 함 SELECT COUNT(1) FROM ( SELECT DISTINCT userId, sessionId, ts, channel FROM adhoc.keeyong_session_summary ); ```
• CTE를 사용해서 중복 제거 후 카운트 해보기 ```sql
• • CTE : FROM 안에 넣지 않고 외부로 빼서 사용 (재사용 가능) With ds AS ( SELECT DISTINCT userId, sessionId, ts, channel FROM adhoc.keeyong_session_summary ) SELECT COUNT(1) FROM ds;

    #### 2. 최근 데이터의 존재 여부 체크하기 (freshness)
    ```sql
    SELECT MIN(ts), MAX(ts)
    FROM adhoc.keeyong_session_summary;

    3. Primary key uniqueness가 지켜지는지 체크하기

    ```sql
• • 1보다 크면 중복이 있다 SELECT sessionId, COUNT(1) FROM adhoc.keeyong_session_summary GROUP BY 1 ORDER BY 2 DESC LIMIT 1;

    #### 4. 값이 비어있는 컬럼들이 있는지 체크하기
    ```sql
    SELECT
    COUNT(CASE WHEN sessionId is NULL THEN 1 END) sessionid_null_count,
    COUNT(CASE WHEN userId is NULL THEN 1 END) userid_null_count,
    COUNT(CASE WHEN ts is NULL THEN 1 END) ts_null_count,
    COUNT(CASE WHEN channel is NULL THEN 1 END) channel_null_count
    FROM adhoc.keeyong_session_summary;

사용자 ID : 보통 웹서비스에서는 등록된 사용자마다 부여하는 유일한 ID 세션 ID : 세션마다 부여되는 ID

• 세션: 사용자의 방문을 논리적인 단위로 나눈 것
  • 사용자가 외부 링크(보통 광고)를 타고 오거나 직접 방문해서 올 경우 세션을 생성
  • 사용자가 방문 후 30분간 interaction이 없다가 뭔가를 하는 경우 새로 세션을 생성
• 즉 하나의 사용자는 여러 개의 세션을 가질 수 있음
• 보통 세션의 경우 세션을 만들어낸 접점(경유지)를 채널이란 이름으로 기록해둠
  • 마케팅 관련 기여도 분석을 위함 (Marketing Channel Attribution)
• 또한 세션이 생긴 시간도 기록

  이 정보를 기반으로 다양한 데이터 분석과 지표 설정이 가능

• 마케팅 관련, 사용자 트래픽 관련
• DAU, WAU, MAU등의 일주월별 Active User 차트
• Marketing Channel Attribution 분석
  • 어느 채널에 광고를 하는 것이 가장 효과적인가?

    데이터베이스와 테이블 예제

    

    SQL 기본

• 먼저 다수의 SQL문을 실행한다면 세미콜론으로 분리 필요
  • SQL문1; SQl문2; SQL문3;
• SQL 주석 ```
• • : 인라인 한줄짜리 주석. 자바에서 //에 해당 /--/: 여러 줄에 걸쳐 사용 가능한 주석 ```
• SQL 키워드는 대문자를 사용한다던지 하는 나름대로 포맷팅이 필요
  • 팀 프로젝트라면 팀에서 사용하는 공통 포맷이 필요
• 테이블/필드이름의 명명규칙을 정하는 것이 중요
  • 단수형 vs. 복수형 - User vs. Users
  • _ vs. CamelCasing
    • user_session_channel vs. UserSessionChannel

SQL DDL - 테이블 구조 정의 언어

CREATE TABLE

• Primary key 속성을 지정할 수 있으나 무시됨
  • Primary key uniqueness
    • Big Data 데디터웨어하우스에서는 지켜지지 않음(Redshift, Snowflake, BigQuery)
• CTAS: CREATE TABLE tabel_name AS SELECT
  • vs. CREATE TABLE and then INSERT

    CREATE TABLE raw_data.user_session_channel (
    userid int,
    sessionid varchar(32) primary key,
    channel varchar(32)
    );

    DROP TABLE

• DROP TABLE table_name;
  • 없는 테이블을 지우려고 하는 경우 에러를 냄
• DROP TABLE IF EXISTS table_name;
• vs. DELETE FROM
  • DELETE FROM은 조건에 맞는 레코드들을 지움 (테이블 자체는 존재)

    ALTER TABLE

• 새로운 컬럼 추가:
  • ALTER TABLE 테이블이름 ADD COLUMN 필드이름 필드타입;
• 기존 컬럼 이름 변경:
  • ALTER TABLE 테이블이름 RENAME 현재필드이름 to 새필드이름
• 기존 컬럼 제거:
  • ALTER TABLE 테이블이름 DROP COLUMN 필드이름;
• 테이블 이름 변경:
  • ALTER TABLE 현재테이블이름 RENAME to 새테이블이름;

SQL DML - 테이블 데이터 조작 언어

레코드 질의 언어: SELECT

• 뒤에서 더 자세히 설명
• SELECT FROM: 테이블에서 레코드와 필드를 읽어오는데 사용
• WHERE를 사용해서 레코드 선택 조건을 지정
• GROUP BY를 통해 정보를 그룹 레벨에서 뽑는데 사용하기도 함
  • DAU, WAU, MAU 계산은 GROUP BY를 필요로 함
• ORDER BY를 사용해서 레코드 순서를 결정하기도 함
• 보통 다수의 테이블의 조인해서 사용하기도 함

  레코드 수정 언어:

• INSERT INTO: 테이블에 레코드를 추가하는데 사용
• UDATE FROM: 테이블 레코드의 필드 값 수정
• DELETE FROM: 테이블에서 레코드를 삭제
  • vs. TRUNCATE (Transaction에 사용 가능 Delete는 불가능)

데이터 분석하기 전 기억할 점

현업에서 깨끗한 데이터란 존재하지 않음

• 항상 데이터를 믿을 수 있는지 의심할 껏! -> 의(疑)데이터증
• 실제 레코드를 몇 개 살펴보는 것 만한 것이 없음 -> 노가다

  데이터 일을 한다면 항상 데이터의 품질을 의심하고 체크하는 버릇이 필요

• 중복된 레코드들 체크하기
• 최근 데이터의 존재 여부 체크하기 (freshness)
• Primary key uniqueness가 지켜지는지 체크하기
• 값이 비어있는 컬럼들이 있는지 체크하기
• 위의 체크는 코딩의 unit test 형태로 만들어 매번 쉽게 체크해볼 수 있음

  어느 시점이 되면 너무나 많은 테이블들이 존재하게 됨

• 회사 성장과 밀접한 관련
• 중요 테이블들이 무엇이고 그것들의 메타 정보를 잘 관리하는 것이 중요해짐

  그 시점부터는 Data Discovery 문제들이 생겨남

• 무슨 테이블에 내가 원하고 신뢰할 수 있는 정보가 들어있나?
• 테이블에 대해 질문을 하고 싶은데 누구에게 질문을 해야하나?

  이 문제를 해결하기 위한 다양한 오픈소스와 서비스들이 출현

• DataHub (LinkedIn), Amundsen (Lyft), ...
• Select Star, DataFrame, …

SELECT

• 테이블(들)에서 레코드들(혹은 레코드수)을 읽어오는데 사용
• WHERE를 사용해 조건을 만족하는 레코드

  SELECT 필드이름1, 필드이름2, …
  FROM 테이블이름
  WHERE 선택조건
  GROUP BY 필드이름1, 필드이름2, ...
  ORDER BY 필드이름 [ASC|DESC] -- 필드 이름 대신에 숫자 사용 가능
  LIMIT N;

  ```sql
• • 예제 SELECT * FROM raw_data.user_session_channel;

SELECT userId, sessionId, channel FROM raw_data.user_session_channel;

SELECT * FROM raw_data.user_session_channel LIMIT 10;

SELECT DISTINCT channel -- 유일한 채널 이름을 알고 싶은 경우 FROM raw_data.user_session_channel;

SELECT channel, COUNT(1) -- 채널별 카운트를 하고 싶은 경우. COUNT 함수!! FROM raw_data.user_session_channel GROUP BY 1;

SELECT COUNT(1) -- 테이블의 모든 레코드 수 카운트. COUNT(*). 하나의 레코드 FROM raw_data.user_session_channel;

SELECT COUNT(1) FROM raw_data.user_session_channel WHERE channel = 'Facebook'; -- channel 이름이 Facebook경우만 고려해서 레코드수 카운트

### CASE WHEN
- 필드 값의 변환을 위해 사용 가능
    - CASE WHEN 조건 THEN 참일때 값 ELSE 거짓일때 값 END 필드이름
- 여러 조건을 사용하여 변환하는 것도 가능
```sql
SELECT CASE
 WHEN channel in ('Facebook', 'Instagram') THEN 'Social-Media'
 WHEN channel in ('Google', 'Naver') THEN 'Search-Engine'
 ELSE 'Something-Else'
END channel_type
FROM raw_data.user_session_channel;

NULL이란?

• 값이 존재하지 않음을 나타내는 상수. 0 혹은 ""과는 다름
• 필드 지정시 값이 없는 경우 NULL로 지정 가능
  • 테이블 정의시 디폴트 값으로도 지정 가능
• 어떤 필드의 값이 NULL인지 아닌지 비교는 특수한 문법을 필요로 함
  • field1 is NULL 혹은 field1 is not NULL
• NULL이 사칙연산에 사용되면 그 결과는?
  • SELECT 0 + NULL, 0 - NULL, 0 * NULL, 0/NULL

COUNT 함수 제대로 이해하기

• 예제 ppt SELECT COUNT(NULL) FROM count_test -> 0 SELECT COUNT(value) FROM count_test -> 6 SELECT COUNT(DISTINCT value) FROM count_test -> 4

WHERE

IN

• WHERE channel in (‘Google’, ‘Youtube’)
  • WHERE channel = ‘Google’ OR channel = ‘Youtube’
• NOT IN

  LIKE and ILIKE

• LIKE is a case sensitive string match. ILIKE is a case-insensitive string match
• WHERE channel LIKE ‘G%’ -> ‘G*’
• WHERE channel LIKE ‘%o%’ -> ‘o’
• NOT LIKE or NOT ILIKE

  BETWEEN

• Used for date range matching

  위의 오퍼레이터들은 CASE WHEN 사이에서도 사용가능

STRING Functions

SQL 연습:

SELECT
 LEN(channel), 
 UPPER(channel),
 LOWER(channel), 
 LEFT(channel, 4)
FROM raw_data.user_session_channel;

• LPAD, RPAD
• SUBSTRING

ORDER BY

• Default ordering is ascending
  • ORDER BY 1 ASC
• Descending requires “DESC”
  • ORDER BY 1 DESC
• Ordering by multiple columns:
  • ORDER BY 1 DESC, 2, 3
• NULL 값 순서는?
  • NULL 값들은 오름차순 일 경우 (ASC), 마지막에 위치함
  • NULL 값들은 내림차순 일 경우 (DESC) 처음에 위치함
  • 이를 바꾸고 싶다면 NULLS FIRST 혹은 NULLS LAST를 사용

타입 변환

DATE Conversion:

• 타임존 관련 변환
  • CONVERT_TIMEZONE('America/Los_Angeles', ts)
  • select pg_timezone_names();
• DATE, TRUNCATE
• DATE_TRUNC
  • 첫번째 인자가 어떤 값을 추출하는지 지정 (week, month, day, …)
• EXTRACT or DATE_PART: 날짜시간에서 특정 부분의 값을 추출가능
• DATEDIFF
• DATEADD
• GET_CURRENT, ...

  TO_CHAR, TO_TIMESTAMP

  Type Casting

  1/2의 결과는?

• 0이 됨. 정수간의 연산은 정수가 되어야하기 때문
  • 분자나 분모 중의 하나를 float로 캐스팅해야 0.5가 나옴
  • 이는 프로그래밍 언어에서도 일반적으로 동일하게 동작함
• 뒤에서 예제를 살펴볼 예정

  :: 오퍼레이터를 사용

• category::float

  cast 함수를 사용

• cast(category as float)