apache-kafka - 如何处理一批最大大小的 KStream 或回退到时间窗口?
问题描述
我想创建一个基于 Kafka 流的应用程序,它处理一个主题并以大小为 X(即 50)的批次接收消息,但如果流的流量较低,则在 Y 秒内(即 5)给我流的任何内容。
因此,我不是一一处理消息,而是处理List[Record]
列表大小为 50(或者可能更小)的 a 。
这是为了使一些 I/O 绑定处理更有效。
我知道这可以用经典的 Kafka API 来实现,但正在寻找一种基于流的实现,它也可以在本地处理偏移提交,同时考虑到错误/失败。我在他的文档中或通过四处搜索找不到任何相关的东西,我想知道是否有人能解决这个问题。
解决方案
@Matthias J. Sax 的回答很好,我只想为此添加一个示例,我认为它可能对某人有用。假设我们想将传入的值组合成以下类型:
public class MultipleValues { private List<String> values; }
要将消息收集到最大大小的批次中,我们需要创建转换器:
public class MultipleValuesTransformer implements Transformer<String, String, KeyValue<String, MultipleValues>> {
private ProcessorContext processorContext;
private String stateStoreName;
private KeyValueStore<String, MultipleValues> keyValueStore;
private Cancellable scheduledPunctuator;
public MultipleValuesTransformer(String stateStoreName) {
this.stateStoreName = stateStoreName;
}
@Override
public void init(ProcessorContext processorContext) {
this.processorContext = processorContext;
this.keyValueStore = (KeyValueStore) processorContext.getStateStore(stateStoreName);
scheduledPunctuator = processorContext.schedule(Duration.ofSeconds(30), PunctuationType.WALL_CLOCK_TIME, this::doPunctuate);
}
@Override
public KeyValue<String, MultipleValues> transform(String key, String value) {
MultipleValues itemValueFromStore = keyValueStore.get(key);
if (isNull(itemValueFromStore)) {
itemValueFromStore = MultipleValues.builder().values(Collections.singletonList(value)).build();
} else {
List<String> values = new ArrayList<>(itemValueFromStore.getValues());
values.add(value);
itemValueFromStore = itemValueFromStore.toBuilder()
.values(values)
.build();
}
if (itemValueFromStore.getValues().size() >= 50) {
processorContext.forward(key, itemValueFromStore);
keyValueStore.put(key, null);
} else {
keyValueStore.put(key, itemValueFromStore);
}
return null;
}
private void doPunctuate(long timestamp) {
KeyValueIterator<String, MultipleValues> valuesIterator = keyValueStore.all();
while (valuesIterator.hasNext()) {
KeyValue<String, MultipleValues> keyValue = valuesIterator.next();
if (nonNull(keyValue.value)) {
processorContext.forward(keyValue.key, keyValue.value);
keyValueStore.put(keyValue.key, null);
}
}
}
@Override
public void close() {
scheduledPunctuator.cancel();
}
}
我们需要创建键值存储,将其添加到,并使用方法StreamsBuilder
构建KStream
流程transform
Properties props = new Properties();
...
Serde<MultipleValues> multipleValuesSerge = Serdes.serdeFrom(new JsonSerializer<>(), new JsonDeserializer<>(MultipleValues.class));
StreamsBuilder builder = new StreamsBuilder();
String storeName = "multipleValuesStore";
KeyValueBytesStoreSupplier storeSupplier = Stores.persistentKeyValueStore(storeName);
StoreBuilder<KeyValueStore<String, MultipleValues>> storeBuilder =
Stores.keyValueStoreBuilder(storeSupplier, Serdes.String(), multipleValuesSerge);
builder.addStateStore(storeBuilder);
builder.stream("source", Consumed.with(Serdes.String(), Serdes.String()))
.transform(() -> new MultipleValuesTransformer(storeName), storeName)
.print(Printed.<String, MultipleValues>toSysOut().withLabel("transformedMultipleValues"));
KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), props);
kafkaStreams.start();
通过这种方法,我们使用了我们进行聚合的传入键。如果您需要通过某些消息的字段而不是键来收集消息,则需要以下流程来触发 KStream 上的重新平衡(通过使用中间主题):
.selectKey(..)
.through(intermediateTopicName)
.transform( ..)
推荐阅读
- ruby-on-rails - Conditionally Omit/Hide Button in Rails
- python - 跨平台 CRC8 函数:C++ 和 Python 奇偶校验
- python - Issues with PySpark on Jupyter notebook with .show() DataFrame once it is passed through a function
- r - 使用不断变化的列表重命名 R 数据框中的列的函数
- python - InvalidArgumentError: input_1_1:0 is both fed and fetched
- sql-server - 使用 sp_send_dbmail 发送的邮件将嵌入的图像显示为附件
- vba - 公式中的Excel VBA变量
- r - Eigenvalues of a matrix, assuming symmetry
- php - 获取解析错误,PHP代码中的语法错误
- python - 获取列的“循环滞后”