Transaction

멱등성 (Impodent)

간단한 ETL

실습 ETL 개요

ETL 추가 문제

실습 Python ETL 개선하기

트랜잭션이란?

멱등성 (Impodent)

멱등성(Impodent)이란 무엇인가?

•

데이터 파이프라인이 연속 실행되었을 때 소스에 있는 데이터가 그대로 DW로
저장되어야함을 이야기

a. No duplicates, no missing data

◦

만약 Incremental Update 시, 중복을 제거하지 않고 계속 적재하는 소스 코드라면, 멱등성이 지켜지지 않은 것.

•

Full refresh를 하는 데이터 파이프라인이라면?

1. 먼저 DW의 관련 테이블에서 모든 레코드를 삭제 2. 데이터 소스에서 읽어온 데이터를 DW 테이블로 적재

◦

만일 1번이 성공하고 2번이 실패한다면? 멱등성 깨짐 ⇒ 원본 데이터와 DW 테이블이 일치 X, 정합성이 깨짐

◦

만일 1번이 실행된 다음에 누군가 이 테이블을 사용한다면?

간단한 ETL

Extract, Transform, Load

•

Extract:

◦

데이터를 데이터 소스에서 읽어내는 과정

•

Transform:

◦

필요하다면 그 원본 데이터의 포맷을 원하는 형태로 변경시키는 과정. 굳이 변환할 필요는 없다

•

Load:

◦

최종적으로 Data Warehouse에 테이블로 집어넣는 과정

•

하나의 DAG는 여러 개의 task라는 단위로 나누어져 있음

실습 ETL 개요

•

웹(S3) 상에 존재하는 이름, 성별 내용의 CSV 파일을 다운받아서 Redshift에 있는 테이블로 복사

각자에게 할당된 schema밑에 아래 테이블을 생성

CREATE TABLE (본인의스키마).name_gender (
 name varchar(32) primary key,
 gender varchar(8)
);
SQL
복사

•

하지만 데이터웨어하우스는 PK를 보장해주지 않는다. 암묵적 룰로 보장을 해주어야한다.

데이터 소스

S3 내부 csv 파일에는 2개의 필드가 존재 (name, gender)

파이썬으로 작성: 세 개의 함수로 구성

extract, transform, load

◦

위의 과정은 재개하면 중복이 생기게 된다

▪

멱등성을 보장해주어야 한다

◦

또한 한 줄씩 업데이트 하는 게 아니라 파일 배치성으로 bulk update를 해주어야 한다

◦

한 줄씩 인서트하면서 pk 확인을 해주게 되면 효율이 안나온다

ETL 추가 문제

헤더가 레코드로 추가되는 문제 해결하기

Idempotent하게 잡을 만들기 (full refresh 잡이라고 가정)
a. 여러 번 실행해도 동일한 결과가 나오게 만들기

Transaction을 사용해보기

•

BEGIN; DELETE FROM ..; 
INSERT INTO… ;END;

정답 코드

실습 Python ETL 개선하기

•

데이터 웨어하우스에서 테이블 업데이트 방법은 크게 두 가지

◦

Full Refresh

▪

단순해서 좋지만 데이터가 커지면 사용불가능

▪

위 과제는 Full Refresh하는 예라고 보면 됨

◦

Incremental Update

▪

데이터가 클 경우 효과적이지만 복잡도 증가

▪

보통 타임스탬프 혹은 일련 번호 등의 필드 필요

▪

execution_date 활용

•

DELETE FROM vs. TRUNCATE

◦

둘 다 테이블에서 레코드를 삭제시켜주는 것

◦

DELETE FROM 명령

▪

테이블에서 특정 레코드를 삭제하는 데 사용됩니다.

▪

Transaction을 존중한다

◦

TRUNCATE 명령

▪

테이블에서 전체 데이터를 삭제하는 데 사용됩니다.

▪

큰 테이블을 삭제할 때 DELETE FROM보다 빠르다.

▪

Transaction을 무시하고 삭제를 해버린다

•

함부로 쓰면 문제가 생길 여지가 있다.

트랜잭션이란?

•

트랜잭션이란?

◦

중간에 실패하면 불완전 상황에 놓이는 작업들이 있다면?

◦

아래 예에서 인출은 성공했지만 송금에서 문제가 생긴다면?

•

트랜잭션 정의

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

◦

SQL, 관계형 DB에서 사용되는 개념

◦

https://postgresql.kr/docs/9.2/tutorial-transactions.html

◦

BEGIN과 END 혹은 BEGIN과 COMMIT 사이에 해당 SQL들을 배치 및 ㅌㅈ사용

◦

COMMIT을 통해 Pysical 테이블에 반영을 시킨다.

◦

Transaction 사이에 정합성이 깨질 경우, ROLLBACK 실행 ⇒ Transaction 이전 상태로 돌아간다

•

Redshift(postgre DB) 연결 함수

import psycopg2

# postgresql db를 연결 - redshift
# Redshift connection 함수
# 본인 ID/PW 사용!
def get_Redshift_connection(boolean=True):
    host = ""
    redshift_user = ""
    redshift_pass = ""
    port = 5439
    dbname = ""
    conn = psycopg2.connect("dbname={dbname} user={user} host={host} password={password} port={port}".format(
        dbname=dbname,
        user=redshift_user,
        password=redshift_pass,
        host=host,
        port=port
    ))
    conn.set_session(autocommit=boolean)
    return conn.cursor()
Python
복사

•

두 가지 종류의 트랜잭션이 존재

◦

conn.set_session(autocommit=boolean)

◦

레코드 변경을 바로 반영하는지 여부. autocommit이라는 파라미터로 조절가능

◦

autocommit 은 쓰기 및 삭제 작업일 때 중요함

•

autocommit=True

◦

기본적으로 별도의 트랜잭션 구간이 없다면 모든 SQL statement가 바로 커밋됨

◦

이를 바꾸고 싶다면 BEGIN;END; 혹은 BEGIN;COMMIT을 사용 (혹은 ROLLBACK)

◦

COMMIT 전까지는 스테이징 테이블에만 보이게 됩니다

•

autocommit=False

◦

기본적으로 모든 SQL statement가 커밋되지 않음

◦

Commit 전까지는 작성 중인 사용자에게만 보인다. (물리적 DB에 반영이 되지 않는다)

◦

커넥션 객체의 .commit()과 .rollback()함수로 커밋할지 말지 결정

•

2개 중에서 무엇을 사용할 지는 개인 취향

◦

Python의 경우 try/catch와 같이 사용하는 것이 일반적

▪

try/catch로 에러가 나면 rollback을 명시적으로 실행. 에러가 안 나면 commit을 실행

conn = get_Redshift_connection(True)
# True 일 때는 바로 physical table에 작성 및 반영을 해버린다.
cur = conn.cursor()

cur.execute("DELETE FROM hajuny129.name_gender;")

try:
  cur.execute("BEGIN;")
  cur.execute("DELETE FROM hajuny129.name_gender;") 
  cur.execute("INSERT INTO hajuny129.name_gender VALUES ('Claire', 'Female');")
  cur.execute("END;")
except (Exception, psycopg2.DatabaseError) as error:
  print(error)
  cur.execute("ROLLBACK;")
# finally :
#   conn.close()
Python
복사

◦

try/except 사용시 유의할 점

▪

아래와 같은 코드가 있었다면 exception이 발생한 경우 어떻게 될까?

try:
  cur.execute(create_sql)
  cur.execute("COMMIT;")
except Exception as e:
  cur.execute("ROLLBACK;")
  raise # 에러 전파 => transaction fail
Python
복사

▪

만약 raise 문이 없다면 에러가 발생했다는 사실이 외부로 전파가 되지 않습니다. 마치 에러가 발생했는데, 아무 일도 없던 것처럼 인지될 수 있습니다.

▪

except에서 raise를 호출하면 발생한 원래 exception이 위로 전파됩니다.

•

에러가 전파된 후, 트랜잭션이 Fail될 것입니다.

•

ETL을 관리하는 입장에서 어떤 에러가 감춰지는 것보다는 명확하게 드러나는 것이 더 좋음

•

위의 경우 cur.execute 뒤에 raise를 호출하는 것이 좋음

•

간단한 실습:

◦

Transaction 배우기

Programmers_DE_Start/Transaction_배우기.ipynb at main · HaJunYoo/Programmers_DE_Start

You can't perform that action at this time. You signed in with another tab or window. You signed out in another tab or window. Reload to refresh your session. Reload to refresh your session.

https://github.com/HaJunYoo/Programmers_DE_Start/blob/main/Transaction_%EB%B0%B0%EC%9A%B0%EA%B8%B0.ipynb