如何管理Kafka KStream到Kstream窗口加入？

有关详细信息,请参阅http://docs.confluent.io/current/streams/developer-guide.html#joining-streams.

滑动窗口基本上定义了一个额外的连接谓词.在类似SQL的语法中,这将是:

SELECT * FROM stream1, stream2
WHERE
   stream1.key = stream2.key
   AND
   stream1.ts - before <= stream2.ts
   AND
   stream2.ts <= stream1.ts + after

其中before == after == joinWindowSizeMs在本实施例中.如果您使用和显式设置这些值before,after也可以使用不同的值.JoinWindows#before()JoinWindows#after()

源主题的保留时间完全独立于windowRetentionTimeMs应用于Kafka Streams自身创建的更改日志主题的指定.窗口保留允许彼此连接无序记录,即迟到的记录(请记住,Kafka具有基于偏移的订购保证,但是关于时间戳,记录可能是乱序的) .

除了Matthias J.Sax所说的之外，还有一个流对流（窗口式）连接示例，位于：https : //github.com/confluentinc/examples/blob/3.1.x/kafka-streams/src/test /java/io/confluent/examples/streams/StreamToStreamJoinIntegrationTest.java

这适用于带有Apache Kafka 0.10.1的Confluent 3.1.x，即截至2017年1月的最新版本。有关master使用较新版本的代码示例，请参见上面的存储库中的分支。

这是上面代码示例的关键部分（同样，对于Kafka 0.10.1），略微适合您的问题。请注意，此示例恰好演示了OUTER JOIN。

long joinWindowSizeMs = TimeUnit.MINUTES.toMillis(5);
long windowRetentionTimeMs = TimeUnit.DAYS.toMillis(30);

final Serde stringSerde = Serdes.String();
KStreamBuilder builder = new KStreamBuilder();
KStream alerts = builder.stream(stringSerde, stringSerde, "adImpressionsTopic");
KStream incidents = builder.stream(stringSerde, stringSerde, "adClicksTopic");

KStream impressionsAndClicks = alerts.outerJoin(incidents,
    (impressionValue, clickValue) -> impressionValue + "/" + clickValue,
    // KStream-KStream joins are always windowed joins, hence we must provide a join window.
    JoinWindows.of(joinWindowSizeMs).until(windowRetentionTimeMs),
    stringSerde, stringSerde, stringSerde);

// Write the results to the output topic.
impressionsAndClicks.to(stringSerde, stringSerde, "outputTopic");