StreamsBuilder
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream = builder.stream("input-topic");
KTable<String, String> table = builder.table("input-topic");
GlobalKTable<String, String> global = builder.globalTable("input-topic");
Topology topology = builder.build();
| 方法 | 输出 | 说明 |
|---|
stream(topic) | KStream | 记录流 |
table(topic) | KTable | 更新流 |
globalTable(topic) | GlobalKTable | 全局表 |
KStream 操作
// 无状态
stream.filter((k, v) -> v.length() > 0)
.map((k, v) -> KeyValue.pair(k, v.toUpperCase()))
.mapValues(String::toUpperCase)
.flatMapValues(v -> Arrays.asList(v.split(" ")))
.selectKey((k, v) -> v.substring(0, 3));
// 有状态
stream.groupByKey()
.count(Materialized.as("count-store"));
stream.groupByKey()
.reduce((a, b) -> a + b, Materialized.as("reduce-store"));
stream.groupByKey()
.aggregate(
() -> 0.0,
(key, value, agg) -> agg + Double.parseDouble(value),
Materialized.as("agg-store")
);
| 操作 | 输入 | 输出 | 状态 |
|---|
filter | KStream | KStream | 无 |
map | KStream | KStream | 无 |
mapValues | KStream/KTable | KStream/KTable | 无 |
flatMapValues | KStream | KStream | 无 |
selectKey | KStream | KStream | 无（改变 key 后会触发 repartition） |
groupByKey | KStream | KGroupedStream | 无 |
count | KGroupedStream | KTable | 有 |
reduce | KGroupedStream | KTable | 有 |
aggregate | KGroupedStream | KTable | 有 |
map 与 selectKey 都被标记为 key-changing：下游一旦做聚合，Kafka Streams 会插入 repartition topic 把相同 key 拉到同一分区；mapValues / flatMapValues 不改变 key，不触发 repartition。
Join
// KStream + KStream（Windowed Join）
stream1.join(stream2, (v1, v2) -> v1 + "-" + v2,
JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)));
// KStream + KTable（非窗口）
stream.leftJoin(table, (s, t) -> s + ":" + t);
// KTable + KTable
KTable<String, String> joined = table1.join(table2, (v1, v2) -> v1 + v2);
| Join | 类型 | 窗口 |
|---|
stream.join(stream) | KStream + KStream | Windowed |
stream.leftJoin(table) | KStream + KTable | 无 |
stream.join(table) | KStream + KTable | 无 |
table.join(table) | KTable + KTable | 无 |
状态存储
// KeyValueStore
Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore("my-store"),
Serdes.String(), Serdes.Long()
);
// WindowStore
Stores.windowStoreBuilder(
Stores.persistentWindowStore("win-store",
Duration.ofDays(7), Duration.ofHours(1), false),
Serdes.String(), Serdes.Long()
);
| 类型 | API | 持久化 |
|---|
| KeyValueStore | persistentKeyValueStore() | RocksDB |
| WindowStore | persistentWindowStore() | RocksDB |
| InMemory | inMemoryKeyValueStore() | 堆内存 |
配置
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10000);
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);
| 配置 | 说明 |
|---|
APPLICATION_ID_CONFIG | Consumer Group ID |
PROCESSING_GUARANTEE | EXACTLY_ONCE_V2 |
COMMIT_INTERVAL_MS | Offset 提交间隔 |
NUM_STREAM_THREADS | 每个实例的线程数 |
测试
@Test
void test() {
try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
TestInputTopic<String, String> input = driver.createInputTopic(
"input", new StringSerializer(), new StringSerializer());
TestOutputTopic<String, String> output = driver.createOutputTopic(
"output", new StringDeserializer(), new StringDeserializer());
input.pipeInput("key", "value");
assertEquals("VALUE", output.readValue());
}
}
返回 目录
。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。