Các phương thức join() trong Spring for Apache Kafka Streams là các inner join, dùng để join các message có cùng key được publish vào 2 Apache Kafka topics.
Ví dụ như mình có 2 topic là customers và orders. Để có thể join các message có cùng key trong 2 topic này, các bạn có thể tạo mới 2 đối tượng KStream và sử dụng phương thức join() để join các message này trong một khoảng thời gian nào đó, như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
@Bean public KStream<String, String> customers(StreamsBuilder builder) { KStream<String, String> customers = builder.stream("customers"); KStream<String, String> orders = builder.stream("orders"); customers .join( orders, (ValueJoiner<String, String, String>) (customer, order) -> String.format("Customer %s ordered %s", customer, order), JoinWindows.ofTimeDifferenceAndGrace(Duration.ofSeconds(25), Duration.ofSeconds(150)), StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String())) .to("enriched-orders"); return customers; } |
- Tham số thứ nhất là tên topic mà chúng ta muốn join.
- Tham số thứ hai là output mà chúng ta muốn khi nhận được value của các message key trong 2 topic.
- Tham số thứ ba là khoảng thời gian mà các message này được public. Các bạn có thể sử dụng 2 phương thức static của class
JoinWindowslàofTimeDifferenceAndGrace()vàofTimeDifferenceWithNoGrace()để cấu hình cho phần join windows này. Phương thứcofTimeDifferenceAndGrace()sẽ định nghĩa khoảng thời gian mà các message được tạo và nhận trong khoảng thời gian này có cộng thêm một khoảng thời gian trừ hao, sẽ được join. Còn phương thứcofTimeDifferenceWithNoGrace()sẽ chỉ join các message được tạo và nhận trong khoảng thời gian này mà thôi. - Tham số thứ 4 định nghĩa kiểu dữ liệu dành cho việc serialization và deserialization các message đó các bạn.
Chạy ví dụ này và publish 1 message với key là 001, value là Khanh vào topic customers, và một message khác với key là 001 và value là “Table” vào topic orders, các bạn sẽ thấy một message được publish với topic enriched-orders với nội dung như sau:
Các bạn có thể join:
- KStream với KStream
- KStream với KTable
- KStream với GlobalKTable
- KTable với KTable
các bạn nhé!
Nếu các bạn join với KTable hoặc GlobalKTable thì các bạn không cần chỉ định khoảng thời gian.
Các bạn có thể xem thêm ở đây
