Apache Kafka ‐ Join
-
customer_id를 키로 양쪽을 같은 파티션에 정렬(co-partition)해, 짝이 될 레코드가 항상 같은 태스크에 모이게 만든다. - 각 태스크는 한쪽을 로컬 state store에 붙들어두고, 반대쪽 레코드가 도착하면 로컬 조회 → 컬럼 병합 → 새 토픽에 발행을 이벤트 단위로 반복한다.
- 원격 조회를 로컬 조회로 바꾸는 것. 마주칠 데이터를 미리 같은 자리에 모아두고 기억해두는 구조를 위해 Join을 사용한다.
drop table simple_user_table;
drop stream simple_user_stream delete topic;
create stream simple_user_stream
(
user_id integer key,
name varchar,
email varchar
) with (
KAFKA_TOPIC = 'simple_user_topic',
KEY_FORMAT = 'KAFKA',
VALUE_FORMAT ='JSON',
PARTITIONS = 3
);
insert into simple_user_stream(user_id, name, email) values (1, 'John', 'test_email_01@test.domain');
insert into simple_user_stream(user_id, name, email) values (2, 'Merry', 'test_email_02@test.domain');
insert into simple_user_stream(user_id, name, email) values (3, 'Elli', 'test_email_03@test.domain');
insert into simple_user_stream(user_id, name, email) values (4, 'Mike', 'test_email_04@test.domain');
insert into simple_user_stream(user_id, name, email) values (5, 'Tom', 'test_email_05@test.domain');
insert into simple_user_stream(user_id, name, email) values (5, 'Tommy', 'test_email_05@test.domain');
insert into simple_user_stream(user_id, name, email) values (6, 'Michell', 'test_email_06@test.domain');
select * from simple_user_stream;- 조인에 참여하는 Stream/Table은 동일한 Co-Partitioning이 적용되어야 한다.
- 조인 Key Partition 개수가 모두 같아야 한다.
- 조인 Key의 타입이 모두 같아야 한다.
- 조인 Key의 파티션별 값 분포는 서로 동일해야 한다.(동일한 Partitioning 전략이 적용되어야 한다)
- Stream-Stream 조인, Stream-Table 조인, Table-Table 조인별로 특정 제약 조건이 존재한다.
- KSQL의 조인은 여러 제약 조건이 있다.
- 조인키에 대한 제약. 조인 키는 같은 데이터 타입이어야 하며 다른 데이터 타입일 경우 CAST()함수를 이용하여 데이터 타입을 변환한다.
- 조인키는 Stream의 경우 Key로, Table의 경우 Primary Key로 지정되어야 한다.
- 조인키의 파티션 갯수는 서로 같아야 한다.
- 조인키의 파티션별 값 분포는 서로 동일하게 분포되어야 한다.
- Stream과 Stream 조인은
within절로 window 시간을 정해줘야 한다. 전체 stream에 대한 조인이 되지 않는다. - Stream과 Stream, Stream과 table 조인은 pull 쿼리가 지원되지 않으며 Push 쿼리가 지원되지 않는다. 즉, 조인을 하게 되면 이는 rocksdb의 도움이 필요하다. 단, Stream과 Stream 조인을 CSAS로 Mview 생성 후에는
select * from mview로 pull 쿼리도 가능하다.
-- 아래는 pull 쿼리로 수행되지 않음.
select a.*, b.*
from simple_user_stream a
inner join user_activity_stream b on a.user= b.user_id;
-- 아래는 push 쿼리지만 within 절이 없어서 오류
select a.*, b.*
from simple_user_stream a
inner join user_activity_stream b on a.user= b.user_id emit changes;
select a.user_id, a.name, b.*
from simple_user_stream a
inner join user_activity_stream b within 2 hours on a.user_id= b.user_id emit changes;- 스트림이 왼쪽, 테이블이 오른쪽. 뒤집으면 조인이 안 된다.
- 스트림 레코드가 오면 결과 발행, 테이블 레코드가 오면 내부 상태만 갱신되고 출력은 없다.
- 해당 키 삭제. 이것도 조인을 일으키지 않고, 다음 스트림 이벤트가 왔을 때 짝이 없는 형태로 드러난다.
- 테이블이 나중에 바뀌어도 이미 나간 결과는 그대로 나간다.
- 테이블은 사건이 아니라 상태라 비교할 시각이 없다. 즉,
WITHIN사용이 불가능하다.
select b.user_id, b.name, a.*
from user_activity_stream a
inner join simple_user_table b on a.user_id= b.user_id emit changes;
-- 아래는 table을 기준으로 stream을 조인하므로 수행되지 않음.
select a.user_id, a.name, b.*
from simple_user_table a
inner join user_activity_stream b on a.user_id= b.user_id emit changes;
-- 아래는 stream-table 조인 시 within 절을 적용하면 수행되지 않음.
select b.user_id, b.name, a.*
from user_activity_stream a
inner join simple_user_table b within 2 hours on a.user_id= b.user_id emit changes;- KSQLDB는 시간의 흐름에 따른 Event Stream 데이터 처리에 기반한다.
- 조인 데이터는 조인 대상인 Stream 또는 Table의 서로 다른 데이터 생성 시점을 감안해 생성된다.
- 조인되는 두 Stream/Table은 동일한 파티션 개수, 동일한 조인 키 타입, 동일한 파티션 분배 방식을 가져야 정상적인 조인이 가능하다.
- 조인 컬럼이 키가 아닌 경우 스트림은 자동 repartition으로 보정되지만 테이블은 직접 재생성해야 한다.
- 파티션 개수 불일치는 에러로 드러나는 반면, 파티셔너 불일치는 에러 없이 조인이 누락되므로 특히 주의해야 한다.
- 한쪽 키는 INT, 다른 쪽은 VARCHAR인 상황에서 다음과 같이 쓰고 싶어진다.
FROM order_stream o
JOIN customer_table c
ON CAST(o.customer_id AS VARCHAR) = c.customer_id- 타입은 맞춰졌으니 될 것 같지만, 파티션 배치는 그대로이다.
- 파티션 번호는 레코드가 토픽에 적재될 때
hash(원본 키 바이트) % 파티션 수로 이미 확정되어 디스크에 놓여 있다.CAST는 그걸 읽어온 뒤 메모리에서 변환하는 연산이라, 이미 정해진 물리적 위치에는 아무 영향이 없다. - 같은 "1"인데 애초에 다른 파티션에 저장되어 있다. CAST로 논리적 값을 맞춰도 마주칠 자리에 있지 않다는 문제는 그대로 남아있다.
- 스트림 쪽에 CAST를 걸면 동작하지만 비용이 붙는다 : 조인 키가 "키 컬럼"이 아니라 "표현식"이 되므로, ksqlDB가 이를 감지해 자동으로 repartition 토픽을 끼워 넣는다. 변환된 값을 키로 다시 써서 배치를 새로 만드는 것이고, 그래서 조인이 정상 동작한다. 카프카에 한 번 더 쓰고 다시 읽기에 지연이 늘고 토픽이 하나 더 생기며, 그게 쿼리가 사는 동안 영구적으로 유지된다.
- 테이블 쪽에 CAST를 걸면 대부분 거부된다 : ksqlDB는 조인의 테이블 변에 대해 조인 표현식이 그 테이블의 키 컬럼 그 자체일 것을 요구한다. 함수나 CAST를 씌우면 키가 아닌 것이 되고, 테이블은 자동 repatition 대상이 아니라서 해결할 방법이 없어 에러로 막는다.
- Table-Table 조인은 빈번하게 사용되지 않는다. 1 : 1 조인만 사용을 권장하며, M : 1 조인 사용 시에는 매우 주의가 필요하다.
- M : 1 조인 시 KEY_FORMAT이 M쪽 KEY가 JSON 포맷으로 되면서 조인 대상 테이블의 조인 키별 파티션 분배가 서로 달라지고 조인 결과가 제대로 생성되지 않는 문제가 발생한다.
- Table-Table 조인 결과는 어느 한쪽 테이블의 데이터가 추가되더라도 이를 반영하여 조인 데이터가 생성된다.
- Group by CTAS로 생성된 테이블과 Master성 테이블 조인을 하는 경우 Stream과 Table 조인 후, Group by로 변경하는 것이 더 효율적이다.
- 리파티셔닝이 발생해서 코파티셔닝 조건을 충족시킨다.
- 스트림 쪽 : 자동으로 리파티셔닝이 끼어들고 조인이 정상 동작한다.
- 테이블 쪽 : 리파티셔닝이 일어나지 않고 에러가 발생한다.
| 항목 | 영향 |
|---|---|
| 지연 | 쓰기 + 읽기 한 번씩 추가 |
| 처리량 | 네트워크·디스크 I/O 증가 |
| 토픽 | 내부 토픽 하나 추가 생성 및 유지 |
| 순서 |
customer_id 기준으로 재정렬 — 기존 order_id 단위 순서는 의미 없어짐 |