我有 2 个 kafka 主题 - recommendations 和 clicks。第一个主题具有由唯一 ID 键控的推荐对象(称为 recommendationsId)。每个产品都有一个用户可以点击的 URL。

clicks 主题获取通过点击推荐给用户的那些产品 URL 生成的消息。设置如此，这些点击消息也由 recommendationId 键入。

注意

1. 推荐和点击之间的关系是一对多的。一个推荐可能会导致多次点击，但一次点击总是与一个推荐相关联。

2. 每个点击对象都会有一个对应的推荐对象。

3. 点击对象的时间戳会晚于推荐对象。

4. 推荐和相应点击之间的间隔可能是几秒到几天(比如最多 7 天)。

我的目标是使用 Kafka 流连接来连接这两个主题。我不清楚的是我应该使用 KStream x KStream 连接还是 KStream x KTable 连接。

我通过按 recommendations 表加入 clicks 流来实现 KStream x KTable 加入。但是，如果推荐是在加入器启动之前生成的，并且点击是在加入器启动后到达的，我将看不到任何加入的点击-推荐对。

我是否使用了正确的连接？我应该使用 KStream x KStream 加入吗？如果是这样，为了能够加入最多 7 天前的推荐点击，我是否应该将窗口大小设置为 7 天？在这种情况下是否还需要设置“保留”期限？

我执行 KStream x KTable 连接的代码如下。请注意，我已经定义了类 Recommendations 和 Click 及其相应的 serde。点击消息只是普通的 String (url)。此 URL 字符串与 Recommendations 对象结合以创建一个 Click 对象，该对象被发送到 jointTopic。

public static void main(String[] args){
    if(args.length!=4){
      throw new RuntimeException("Expected 3 params: bootstraplist clickTopic recsTopic jointTopic");
    }

    final String booststrapList = args[0];
    final String clicksTopic = args[1];
    final String recsTopic = args[2];
    final String jointTopic = args[3];

    Properties config = new Properties();
    config.put(StreamsConfig.APPLICATION_ID_CONFIG, "my_joiner_id");
    config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, booststrapList);
    config.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    config.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, JoinSerdes.CLICK_SERDE.getClass().getName());

    KStreamBuilder builder = new KStreamBuilder();

    // load clicks as KStream
    KStream<String, String> clicksStream = builder.stream(Serdes.String(), Serdes.String(), clicksTopic);

    // load recommendations as KTable
    KTable<String, Recommendations> recsTable = builder.table(Serdes.String(), JoinSerdes.RECS_SERDE, recsTopic);

    // join the two
    KStream<String, Click> join = clicksStream.leftJoin(recsTable, (click, recs) -> new Click(click, recs));

    // emit the join to the jointTopic
    join.to(Serdes.String(), JoinSerdes.CLICK_SERDE, jointTopic);

    // let the action begin
    KafkaStreams streams = new KafkaStreams(builder, config);
    streams.start();
  }

只要在加入器(上述程序)运行之后生成推荐和点击，这就可以正常工作。但是，如果在加入器运行之前生成推荐的点击到达，我看不到任何加入发生。我该如何解决这个问题？

如果解决方案是使用 KStream x KSTream 加入，那么请帮助我了解我应该选择什么窗口大小以及选择什么保留期。

最佳答案

您的整体观察是正确的。从概念上讲，您可以通过两种方式获得正确的结果。如果你使用流表连接，你有两个缺点(这可能会在未来的 Kafka 版本中重新考虑和改进)

• 您已经提到，如果点击在相应推荐之前得到处理，(内部)连接将失败。但是，如您所知会有推荐，您可以使用左连接而不是内部连接，检查连接结果，如果推荐是 null，则将点击事件写回输入主题。 (即，您获得重试逻辑)——或者当然，对单个推荐的连续点击可能会乱序，您可能需要在应用程序代码中考虑到这一点。
• KTable 的第二个缺点它将随着时间的推移永远无限地增长，因为您将向它添加越来越多的独特建议。因此，您需要通过发送 <recommendationsId, null> 形式的墓碑记录来实现一些“过期逻辑”转到推荐主题以删除您不再关心的旧推荐。
• 这种方法的优点是，与流-流连接相比，您总共需要更少的内存/磁盘空间，因为您只需要缓冲应用程序中的所有推荐(但不需要点击)。

如果您使用流-流连接，并且点击可以在推荐后 7 天发生，则您的窗口大小必须为 7 天——否则，点击不会与推荐一起连接。

• 这种方法的缺点是，您将需要更多的内存/磁盘，因为您将缓冲应用程序中过去 7 天的所有点击和所有推荐。
• 优点是顺序或处理(即推荐与点击)不再重要(即您不需要实现上述重试策略)
• 此外，旧的推荐会自动过时，因此您不需要实现特殊的“过期逻辑”。

对于流-流连接，保留时间答案略有不同。它必须至少为 7 天，因为窗口大小为 7 天。否则，您将删除“运行窗口”的记录。您还可以将保留期设置得更长，以便能够处理“迟到的数据”。假设用户在窗口时间范围结束时单击(推荐的 7 天时间跨度结束前 5 分钟)，但该单击仅在 1 小时后报告给您的应用程序。如果您的保留期为 7 天作为您的窗口大小，则无法再处理此迟到的记录(因为建议已经被删除)。如果您设置更长的保留期，例如 8 天，您仍然可以处理延迟记录。这取决于您的应用程序/语义需要您想要使用的保留时间。

总结: 从实现的角度来看，使用流-流连接比使用流-表连接更简单。但是，内存/磁盘节省是预期的，并且可能会很大，具体取决于您的点击流数据速率。

关于java - Kafka流加入，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/46412516/