聚合＆amp;与Kafka流分组

Question

我正在Kafka流和州商店工作。我有以下逻辑，可以根据密钥汇总消息组。

 final Materialized<String, Sample, SessionStore<Bytes, byte[]>> abcStore =
            Materialized.<String, Sample, SessionStore<Bytes, 
 byte[]>>as("topicforchangelog").withCachingDisabled();


 KStream<String, Sample> abcStream = builder.stream("inprocess-topic");


 SessionWindowedKStream<String, Sample> windowedKStream =
            abcStream.groupByKey().windowedBy(SessionWindows.with(ofSeconds("36000")));


    windowedKStream
            .aggregate(
                    Aggregater::initialize,
                    Aggregater::merge,
                    Aggregater::merge,
                    abcStore)
            .toStream()
            .map((key, value) -> KeyValue.pair(key.key(), value))
            .filter((k, v) -> v != null )
            .selectKey((key, value) -> value.getId())
            .peek(abcClass::logOut)
            .to("outputtopic");

基于汇总类中的逻辑，我能够按照我的期望合并以下消息。

 Message 1 : Key : A , value : {"id" : "abc" , "names" : []} 

 Message 2 : Key : A , value : {"id" : null , "names" : [{"name" : "x"}]}

 Message 3 : Key : A , value : {"id" : null , "names" : [{"name" : "xx"}]}

以及我在下面得到的最后一条消息。

 {"id" : "abc" , "names" : [{"name" : "x"}, {"name" : "xx"}]}

首先 三个 消息 是 > 主题 分钟。 但 >（一个 天 或 更多） ， 如果 > i 发送 a 消息 具有 key ”一个“ -

 Message 4 : Key : A , value : {"id" : null ,  "names" : [{"name" : "xxx"}}

我的投影是形成一个类似的信息

{"id" : "abc" , "names" : [{"name" : "x"}, {"name" : "xx"},{"name" : "xxx"}]}

，但没有发生。我正在接收到{“ id”：null，“ names”的输出：[{“ name”：“ xxx”}}}的

聚合并未发生，而所有记录中存在的所有记录都会以“ a”为键

。由于

1. 会话窗口是否已过期
2. 因为消费者的应用程序已经读取了以前的消息？

（或） ，

Answer 1

只是一个疯狂的猜测。因为您正在使用窗口的聚合，所以您可能需要考虑以下内容。

的详细行为

• KgroupedTable/Streame输入记录
• 被忽略了。首次收到记录键时，将调用初始化器（并在加法器和减法器之前调用）。请注意，与KgroupedStream相比，随着时间的推移，由于已接收到该密钥的输入墓碑记录，因此可以多次调用初始化器以不止一次调用（请参阅下文）。
• 当收到一个键（例如，插入）的第一个非NULL值时，则仅调用加法器。
• 当接收到一个键的随后的非挂钩值（例如，更新）时，（1）以旧值为单位存储的旧值和（2）调用旧值，以及（2）将加法器与输入记录的新值调用那是刚刚收到的。未定义减法器和加法器的执行顺序。
• 当收到一个键（例如删除）的墓碑记录（即带有零值的记录）时，仅调用减法器。请注意，每当减法器返回零值本身时，就将相应的键从结果的ktable中删除。如果发生这种情况，该密钥的任何下一个输入记录都会再次触发初始化机。

请参阅表中的第6行。