DEVOCEAN

개요

먼저 ksqlDB의 스트림(Stream)과 테이블(Table) 객체에 대한 간단한 설명입니다

스트림과 테이블은 모두 연속 데이터가 있는 Kafka topic의 래퍼입니다.

스트림은 세상에서 발생하는 이벤트를 캡처한 데이터를 나타내며 다음과 같은 특징이 있습니다.

• 무제한 : 데이터의 끝없는 연속 흐름을 저장하므로 스트림은 제한이 없습니다.

• Immutable : 들어오는 모든 새 데이터는 현재 스트림에 추가하고 기존 레코드를 수정하지 않으며, 변경할 수 없습니다.

테이블은 키를 사용하여 데이터 저장 또는 해당 이벤트 스트림의 구체화된 뷰를 나타내며 다음과 같은 특징을 가지고 있습니다.

• 제한 : 스트림의 스냅샷을 나타냅니다.

• 변경 가능 : 테이블에 동일한 키를 가진 데이터가 들어오는 경우 새 데이터(<Key, Value> 쌍)를 테이블에 추가됩니다. 동일한 키의 데이터가 존재하면 해당 키에 대한 최신 값으로 변경됩니다.

참고: Kafka 주제의 모든 레코드는 키와 값의 쌍으로 표시됩니다. 따라서 테이블에는 항상 주어진 키의 최신값이 표시됩니다.

스트림과 테이블에 대한 설명 예제

거래 및 계정잔액을 활용하여 자금 이체를 관리하려는 은행 시스템을 가정합니다.

Alice와 Bob 두 사용자의 초기 잔액이 각각 200$와 100$라고 가정합니다.

다음은 이 두 사용자 사이에서 발생하는 일련의 자금이체 트랜잭션입니다.

거래 1: 앨리스는 밥에게 100$를 줍니다.

거래 2: 밥은 앨리스에게 50$를 줍니다.

거래 3: 밥은 앨리스에게 100$를 줍니다.

여기서 스트림은 한 계정에서 다른 계정으로 이체된 돈을 기록하는 거래 이벤트의 변경 불가능한 데이터를 나타냅니다.

반면에 테이블은 사용자당 계정의 최신 상태를 반영합니다. 예를 들어 테이블은 현재 잔액을 저장합니다.

테이블은 계정의 최신 상태를 저장하고, 스트림은 트랜잭션 레코드를 저장합니다.

본론 - CDC (Change Data Capture) 연계 방법

이제 본격적으로 원천 Database의 테이블로 부터 ksqlDB의 Table 객체로 CDC 연계하는 방법을 살펴 보겠습니다.

전체 흐름은 아래와 같습니다.

원천 DB (Oracle)의 Table A를 ksqlDB의 Table 객체로 CDC 로 연계하려고 합니다.

1. Kafka Connector로 원천 Database (여기선 Oracle)로 해당 테이블에 대하여 연결

curl --location --request POST 'http://localhost:8083/connectors' \
-H 'Content-Type: application/json' \
-d '{
    "name": "oracle_hub_source_connector"  ,
    "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:oracle:thin:@xx.xx.xx.xx:1521/xxx",
    "connection.user": "xxx",
    "connection.password": "xxx",
    "mode": "timestamp",
    "timestamp.column.name": "xxx",
    "topic.prefix": "ORACLE_",
    "table.whitelist": "Table A",
    "poll.interval.ms": 5000,
    "db.timezone": "Asia/Seoul"
    }}'

2. Stream A 생성

CREATE STREAM STREAM A (
    abc VARCHAR )
WITH (KAFKA_TOPIC='Topic A', VALUE_FORMAT='JSON');

3. Stream A_key 생성

CREATE STREAM STREAM A_key
AS SELECT
   a, b, c
  FROM  STREAM A
    PARTITION BY c
   EMIT CHANGES;

4. Table A_key 생성

CREATE TABLE TABLE A_key(
 a VARCHAR,
 b VARCHAR,
 c VARCHAR)
WITH (KAFKA_TOPIC='Topic A_key', VALUE_FORMAT='JSON', KEY='c');

ksqlDB에 만들어진 Table A_key는 원천 DB의 Table A에 변경 데이터 발생시 즉각적으로 반영되게 된다.