1
[PostgreSQL WAL]
2
LSN 100: INSERT INTO visits (id=1, user='A')
3
LSN 101: UPDATE visits SET user='B' WHERE id=1
4
LSN 102: DELETE FROM visits WHERE id=1
5

6
       ↓ CDC Consumer (Debezium)
7

8
[Kafka Topic: dedp.schema.visits]
9
{op: "c", id: 1, after: {user: "A"}}     ← create
10
{op: "u", id: 1, before: {user: "A"}, after: {user: "B"}}  ← update
11
{op: "d", id: 1, before: {user: "B"}}    ← delete

1
import requests
2

3
debezium_config = {
4
    "name": "visits-connector",
5
    "config": {
6
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
7
        "database.hostname": "postgres",
8
        "database.port": "5432",
9
        "database.user": "postgres",
10
        "database.password": "postgres",
11
        "database.dbname": "postgres",
12
        "database.server.name": "dedp",
13
        "schema.include.list": "dedp_schema",
14
        "topic.prefix": "dedp",
15
        # 아래 설정이 PostgreSQL WAL 연동의 핵심
16
        "plugin.name": "pgoutput",  # logical replication 플러그인
17
    }
18
}
19

20
# Kafka Connect REST API로 커넥터 등록
21
response = requests.post(
22
    "http://kafka-connect:8083/connectors",
23
    json=debezium_config,
24
    headers={"Content-Type": "application/json"}
25
)
26
# 결과: dedp_schema.visits 테이블 → dedp.dedp_schema.visits 토픽으로 CDC 스트리밍

1
from pyspark.sql import functions as F
2

3
orders_stream = spark.readStream.format("kafka").load()
4
users_stream = spark.readStream.format("kafka").load()
5

6
# 스트리밍 JOIN에서는 watermark로 "얼마나 기다릴지" 정의해야 함
7
orders_with_watermark = orders_stream.withWatermark("event_time", "10 minutes")
8
users_with_watermark = users_stream.withWatermark("event_time", "10 minutes")
9

10
joined = orders_with_watermark.join(
11
    users_with_watermark,
12
    on="user_id",
13
    how="inner"  # 한쪽 스트림이 늦으면 매칭이 지연될 수 있음
14
)

1
from pyspark.sql import SparkSession
2
from delta.tables import DeltaTable
3

4
spark = (SparkSession.builder
5
    .config("spark.databricks.delta.properties.defaults.enableChangeDataFeed", "true")
6
    .getOrCreate()
7
)
8

9
# 방법 1: 글로벌 설정 (위의 config)
10
# 방법 2: 테이블 레벨 설정
11
spark.sql("""
12
    CREATE TABLE events (
13
        visit_id STRING, event_time TIMESTAMP, user_id STRING
14
    )
15
    TBLPROPERTIES (delta.enableChangeDataFeed = true)
16
""")
17

18
# CDF 스트리밍 읽기
19
cdf_stream = (
20
    spark.readStream.format("delta")
21
    .option("readChangeFeed", "true")
22
    .option("startingVersion", 0)        # 어떤 버전부터 읽을지
23
    .option("maxFilesPerTrigger", 4)     # 처리량 제어
24
    .table("events")
25
)
26

27
# CDF 출력에는 추가 메타 컬럼이 포함됨:
28
# _change_type: insert | update_preimage | update_postimage | delete
29
# _commit_version: 변경이 발생한 Delta 커밋 버전
30
# _commit_timestamp: 변경 시간
31

32
query = cdf_stream.writeStream.format("console").start()

1
| visit_id | _change_type      | _commit_version |
2
|----------|-------------------|-----------------|
3
| v001     | update_preimage   | 3               |  ← 변경 전
4
| v001     | update_postimage  | 3               |  ← 변경 후

1
[시간축]
2
────────────────────────────────────────────
3
     기존 데이터           CDC 시작 시점 이후
4
  ←── Full Load ──→  ←── CDC 스트리밍 ──→