Kafka - KSQL Join

develkkm·2025년 11월 18일

kafka

0부터 시작하는 Kafka

목록 보기

17/18

ksqlDB는 Kafka Topic으로부터 생성된 Stream 또는 Table 간에 SQL 유사한 JOIN 연산을 수행할 수 있다.

JOIN 연산의 결과는 새로운 Stream 또는 Table이 되며, 두 개 이상의 입력을 공통 키를 기준으로 결합한다.

따라서 실시간 이벤트 스트림이나 상태 테이블을 서로 연결해 다양한 분석 을 구현할 수 있다.

제약사항

JOIN을 성공적으로 수행하기 위해서는 여러 제약사항을 반드시 고려해야 한다.

JOIN 키의 타입 및 스키마가 동일해야 한다.
JOIN 대상 스트림 또는 테이블은 동일한 파티션 수 및 동일한 파티션 전략을 가져야 한다.

Stream to Stream

Stream to Stream JOIN은 두 개 이상의 Stream이 동일 키 또는 동일 조건에서 결합될 때 사용한다.

특징

두 Stream에 각각 동일한 키가 들어오면, 지정된 시간 범위 내에서 서로 매칭되어 결과가 생성된다.
반드시 WITHIN <time> 절을 사용해 윈도우 시간 제약을 설정해야 한다.
OUTER JOIN 형태(INNER, LEFT, RIGHT, FULL)를 지원한다.

CREATE STREAM shipped_orders AS
  SELECT o.id AS orderId,
         o.itemid AS itemId,
         s.id AS shipmentId,
         p.id AS paymentId
  FROM orders o
    INNER JOIN payments p WITHIN 1 HOURS ON p.id = o.id
    INNER JOIN shipments s WITHIN 2 HOURS ON s.id = o.id;

Stream to Table

Stream to Table JOIN은 이벤트 스트림과 상태 테이블을 결합하여 스트림 데이터를 상태 기반으로 보강(enrichment)할 때 사용된다.

특징

Stream은 지속적으로 발생하는 이벤트이며, Table은 키 기준으로 최신 상태를 유지한다.
Stream의 각 레코드가 들어올 때 테이블 조회가 수행되어 결합된 결과가 생성된다.
Stream이 왼쪽(source)여야 하고, 이벤트 생성이 Stream 쪽에서 일어날 때만 JOIN이 트리거된다.
Table 쪽의 업데이트만으로는 JOIN 결과가 자동 생성되지 않는다.

CREATE STREAM pageviews_enriched AS
  SELECT users.userid AS userid,
         pageid,
         regionid,
         gender
  FROM pageviews
    LEFT JOIN users ON pageviews.userid = users.userid
  EMIT CHANGES;

위 예시는 pageviews 스트림 이벤트가 들어올 때 users 테이블을 참조해 유저 정보를 결합하는 방식이다.

Event-Time

Stream to Table JOIN은 언제 JOIN 결과가 생성되는지가 중요하며, ksqlDB는 이를 Stream 이벤트가 도착하는 “이벤트 발생 시점(event time)” 기준으로 처리한다.

JOIN 결과는 Stream 쪽 이벤트가 들어오는 순간에만 생성된다.
Table의 값이 먼저 변경되더라도 Stream 이벤트가 오지 않으면 JOIN 결과는 생성되지 않는다.
즉 Stream은 trigger 역할을 하며, Table은 lookup 역할을 수행한다.
테이블 갱신은 자동으로 JOIN 결과를 발생시키지 않는다.

이 특성 때문에 Stream to Table JOIN은 Enrichment(부가정보 결합) 또는 Stream 이벤트 기반 Lookup 구조로 설명된다.

Table to Table

Table to Table JOIN은 두 개의 상태 테이블을 결합하여 새로운 테이블 형태의 결과를 생성할 때 사용한다.

두 테이블 간 키 기반 결합이 가능하다.
Primary Key 기반(1:1) 또는 Foreign Key 기반(1:N) JOIN을 지원한다.
Many-to-Many( N:M ) JOIN은 지원되지 않는다.

CREATE TABLE orders_by_customer AS
  SELECT c.customer_id,
         count(o.order_id) AS total_orders
  FROM customers c
    INNER JOIN orders o ON c.customer_id = o.customer_id
  GROUP BY c.customer_id
  EMIT CHANGES;