DevCourse TIL Day5 Week6

김태준·2023년 5월 12일

Data Enginnering DevCourse

목록 보기

26/93

이번주차 DW와 SQL 기본 마무리

🎈 사용자 별 처음 채널, 마지막 채널 출력하기

%%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
ORDER BY A.userid;

< 풀이 과정 >
우선, user_session_channel (A) 테이블에 userid, sessionid, channel이 존재하고, session_timstamp (B) 테이블에는 sessionid, ts가 존재한다.
따라서 두 테이블의 sessionid를 기준으로 user_session_channel에 일치하는 sessionid만 매칭을 시키기 위해 LEFT JOIN을 진행한다.
그리고 처음 채널과 마지막 채널을 나타내기 위해 A테이블의 channel의 FIRST_VALUE, LAST_VALUE를 사용해주는데, 다음 과정을 거친다.
1. A테이블의 userid를 기준으로 파티션을 진행하는데, 해당 파티션은 B테이블의 ts필드를 기준으로 오름차순 정렬(오래된 ts부터 나오도록)되어 있다.
2. rows between을 통해 처음 row와 마지막 row의 범위를 현재 행 ~ 윈도우 내 마지막 행 까지로 지정한다.
3. 그렇게 되면 First_Channel은 userid로 기록된 첫번째 channel 명을 갖게 된다.
4. Last_Channel 역시 마찬가지로 userid로 기록된 마지막 channel 명을 갖게 된다.

✅ Transaction

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

데이터의 정합성을 보장하기 위한 기능

논리적인 작업 자체가 ALL(commit) OR NOTHING(ROLLBACK)을 보장하는 것

Atomicity, Consistency, Isolation, Durability를 보장해야 함.

SQL에서 COMMIT 한다는 의미
-> DB에 바로 모든 레코드 작업(수정, 삭제, 추가)이 적용 : autocommit = True
-> 만일 특정 작업을 트랜잭션으로 묶는다면 BEGIN, END(COMMIT)/ROLLBACK으로 처리

✨ Transaction 방식

colab
: 기본적으로 모든 SQL 문이 바로 commit 처리 (autocommit = True)
: 이를 변경하려면 BEGIN:END;, BEGIN:COMMIT;, ROLLBACK사용

psycopg2
: autocommit 파라미터로 조절 가능. True인 경우 PostgreSQL의 commit mode와 동일
: False인 경우 모든 connection object의 .commit(), .rollback()함수로 트랜잭션 조절 가능

connection 과정은 다음과 같다.

import psycopg2

# Redshift connection 함수
def get_Redshift_connection(autocommit):
    host = "learnde.cduaw970ssvt.ap-northeast-2.redshift.amazonaws.com"
    redshift_user = "guest"
    redshift_password = "Guest1!*"
    port = 5439
    dbname = "dev"
    conn = psycopg2.connect("dbname={dbname} user={user} host={host} password={password} port={port}".format(
        dbname = dbname,
        user = redshift_user,
        password = redshift_password,
        host = host,
        port = port
    ))
    conn.set_session(autocommit = autocommit)
    return conn

# insert SQL을 False로 진행
conn = get_Redshift_connection(False)
# 커서 부여 -> 변수 저장
cur = conn.cursor()
# sql문 실행 결과 fetchall 함수로 모든 행 check
cur.execute("SELECT * FROM adhoc.keeyong_name_gender;")
res = cur.fetchall()
for r in res:
	print(r)
# 입력과 동시에 DB 커밋 처리    
cur.execute("INSERT INTO adhoc.keeyong_name_gender VALUES ('Keeyong', 'Male');")
# conn.commit()는 동일한 결과를 가져옴.
cur.execute("COMMIT;")    
# ROLLBACK 원할 경우 아래 코드 실행
cur.execute("ROLLBACK;") 
conn.rollback()

# 아래 셀로 한 번에 실행 처리 가능
try:
  cur.execute("DELETE FROM adhoc.keeyong_name_gender;") 
  cur.execute("INSERT INTO adhoc.keeyong_name_gender VALUES ('Claire', 'Female');")
  conn.commit()
except (Exception, psycopg2.DatabaseError) as error:
  print(error)
  conn.rollback()
finally :
  conn.close()

이전에, 트랜잭션에 대해 다음과 같이 학습을 해보았다.
트랜잭션 고립 수준

✅ DELETE VS TRUNCATE

DELETE FROM TABLE_NAME (NOT DELETE * FROM ~~)

테이블 내 모든 레코드 삭제

WHERE 절로 특정 레코드만 삭제 가능

TRUNCATE

DELETE의 속도가 느린 부분 보완해 전체 테이블 내용 삭제 시에 유리

단, WHERE 절을 지원하지 않고, Transaction 역시 지원하지 않는다.

✅ 기타 문법

UNION : 합집합, 여러 테이블 or SELECT 결과를 중복 제거하여 하나로 보여줌.

UNION ALL : 중복 포함

EXCEPT(MINUS) : 차집합. 하나의 SELECT 결과 - 타 SELECT 결과 처리

INTERSECT : 교집합, 여러 SELECT 문에서 같은 레코드만 찾아 처리

COALESCE(exp1, exp2, exp3) : 첫번째 인자부터 NULL이 아니면 해당 인자 리턴, 인자들이 모두 NULL이면 NULL리턴하는 함수로, 결측치 채울 때 사용

NULLIF(exp1, exp2) : exp1, exp2가 같으면 NULL 리턴

LISTAGG : GROUP BY에서 사용되는 Aggregate함수 중 하나.
ex)
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 userid
LIMIT 10;

# 결과
Youtube->Google->Instagram->Youtube->Instagram->Instagram->Instagram->
WINDOW 함수 (LAG)

userid 기준으로 파티션 진행 후, ts 오름차순 정렬된 상태에서 이전(1) 채널 값들 나열
ex) LAG(channel, 1) OVER (PARTITION BY userId ORDER BY ts) prev_channel

userid 기준으로 파티션 진행 후, ts 내림차순 정렬된 상태에서 이후(1) 채널 값들 나열
ex) LAG(channel, 1) OVER (PARTITION BY userId ORDER BY ts) prev_channel