1
import psycopg2
2
from datetime import datetime
3

4
def demo_read_uncommitted():
5
    """
6
    read uncommitted 컨셉 구현
7
    보통 DB 레벨에서 이런 트랜잭션은 차단됨
8

9
    """
10
    # Connection 1: Writer (Full Load 실행 중)
11
    conn1 = psycopg2.connect("dbname=demo user=etl")
12
    cursor1 = conn1.cursor()
13

14
    print("=" * 60)
15
    print("Read Uncommitted Demo")
16
    print("=" * 60)
17

18
    # Writer: 트랜잭션 시작 커밋안함
19
    cursor1.execute("SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED")
20
    cursor1.execute("BEGIN")
21
    cursor1.execute("DELETE FROM products")
22
    cursor1.execute("INSERT INTO products VALUES (1, 'New Product', 1000)")
23
    print("\n[Writer] 새 데이터 삽입 (아직 커밋 안함)")
24

25
    # Reader: 동시에 데이터 읽기
26
    conn2 = psycopg2.connect("dbname=warehouse user=analyst")
27
    cursor2 = conn2.cursor()
28
    cursor2.execute("SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED")
29
    cursor2.execute("SELECT COUNT(*) FROM products")
30
    count = cursor2.fetchone()[0]
31
    print(f"[Reader] 현재 조회되는 레코드 수: {count}건")
32
    print("아직 커밋 안된 데이터가 보임! (Dirty Read)")
33

34
    # Writer: 롤백
35
    cursor1.execute("ROLLBACK")
36
    print("\n[Writer] 트랜잭션 롤백!")
37

38
    # Reader: 다시 읽기
39
    cursor2.execute("SELECT COUNT(*) FROM products")
40
    count = cursor2.fetchone()[0]
41
    print(f"[Reader] 롤백 후 레코드 수: {count}건")
42

43
    conn1.close()
44
    conn2.close()

1
def demo_read_committed():
2
    """
3
    Dirty Read 방지
4
    """
5

6
    print("=" * 60)
7
    print("Read Committed demo")
8
    print("=" * 60)
9

10
    conn1 = psycopg2.connect("dbname=warehouse user=etl")
11
    cursor1 = conn1.cursor()
12

13
    conn2 = psycopg2.connect("dbname=warehouse user=analyst")
14
    cursor2 = cursor2.cursor()
15

16
    # init
17
    cursor1.execute("DELETE FROM products")
18
    cursor1.execute("INSERT INTO products VALUES (1, 'Product A', 1000)")
19
    conn1.commit()
20
    print("[초기] Product A가 있음")
21

22
    #  Reader: 첫 번째 읽기
23
    cursor2.execute("SET TRANSACTION ISOLATION LEVEL READ COMMITTED")
24
    cursor2.execute("BEGIN")
25
    cursor2.execute("SELECT price FROM products WHERE id = 1")
26
    price1 = cursor2.fetchone()[0]
27
    print(f"\n[Reader] 첫 번째 읽기: 가격 = {price1}원")
28

29
    # Writer: 가격 변경하고 커밋
30
    cursor1.execute("UPDATE products SET price = 2000 WHERE id = 1")
31
    conn1.commit()
32
    print("[Writer] 가격을 2000원으로 변경하고 커밋")
33

34
    # Reader: 같은 트랜잭션 내에서 다시 읽기
35
    cursor2.execute("SELECT price FROM products WHERE id = 1")
36
    price2 = cursor2.fetchone()[0]
37
    print(f"[Reader] 두 번째 읽기: 가격 = {price2}원")
38
    print(f" 같은 트랜잭션인데 값이 {price1}원 → {price2}원으로 변경됨!")
39
    print(" (Non-Repeatable Read)")
40

41
    cursor2.execute("COMMIT")
42

43
    conn1.close()
44
    conn2.close()
45

46
 demo_read_committed()

1
from pyspark.sql import SparkSession
2

3
def full_load_from_postgres_with_isolation():
4
    """
5
    PySpark로 PostgreSQL에서 Full Load 시 Isolation Level 설정
6
    """
7

8
    spark = SparkSession.builder \
9
        .appName("FullLoadWithIsolation") \
10
        .config("spark.jars", "/path/to/postgresql-jdbc.jar") \
11
        .getOrCreate()
12

13
    # Repeatable Read로 설정하여 일관된 스냅샷 읽기
14
    jdbc_url = "jdbc:postgresql://source-db:5432/production"
15

16
    # 옵션 1: JDBC 연결 레벨에서 설정
17
    df = spark.read \
18
        .format("jdbc") \
19
        .option("url", jdbc_url) \
20
        .option("dbtable", "products") \
21
        .option("user", "etl") \
22
        .option("password", "password") \
23
        .option("isolationLevel", "REPEATABLE_READ") \
24
        .load()
25

26
    print(f"{df.count():,}건 읽기 완료 (Repeatable Read)")
27

28
    # 쿼리로 명시적 설정
29
    custom_query = """
30
        (
31
            SELECT set_config('transaction_isolation', 'repeatable read', true);
32
            SELECT * FROM products
33
        ) as subq
34
    """
35

36
    df2 = spark.read \
37
        .format("jdbc") \
38
        .option("url", jdbc_url) \
39
        .option("dbtable", custom_query) \
40
        .option("user", "etl_user") \
41
        .option("password", "password") \
42
        .load()
43

44
    df.write \
45
        .mode("overwrite") \
46
        .parquet("s3://datalake/products/")
47

48
    spark.stop()

1
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;
2
SET TRANSACTION ISOLATION LEVEL REPEATABLE_READ
3
SET TRANSACTION ISOLATION LEVEL READ COMMITTED
4

5
-- READ UNCOMMITTED는 postges의 경우 DB 수준에서 막기 때문에 READ COMMITTED로 대체됨
6
SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED