如何批量处理最大大小的 KStream 或回退到时间窗口？

Question

我想创建一个基于 Kafka 流的应用程序，该应用程序处理一个主题并分批接收大小为 X（即 50）的消息，但如果流的流量较低，则在 Y 秒（即 5）内给我任何流。

因此，我不是一个一个地处理消息，而是处理List[Record]列表大小为 50（或可能更少）的 a。

这是为了使一些 I/O 绑定处理更有效。

我知道这可以用经典的 Kafka API 来实现，但我正在寻找一种基于流的实现，它也可以在本地处理偏移提交，并考虑到错误/失败。我在他的文档中或通过四处搜索都找不到任何相关的内容，并且想知道是否有人可以解决此问题。

Answer 1

@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( ..)