Kafka流连接

Question

我有两个Kafka主题-推荐和点击。第一个主题包含由唯一Id（称为recommendationsId）键入的recommendations对象。每个产品都有一个用户可以单击的URL。

clicks主题获取通过单击推荐给用户的产品URL生成的消息。它是如此设置的，这些单击消息也由推荐id键控。

请注意

>

每个单击对象都会有一个相应的推荐对象。

click对象的时间戳将晚于Recommensions对象。

建议和相应的点击之间的间隔可能是几秒钟到几天（最多7天）。

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

我通过加入clicksstreambyrecommendations表来实现KStream x KTablejoin。但是，如果建议是在joiner启动之前生成的，而单击是在joiner启动之后到达的，则我无法看到任何连接的clicks建议对。

我是否使用了正确的连接？我应该使用KStream x KStream连接吗？如果是的话，为了最多过去7天能够加入一个有推荐的点击，我是不是应该把窗口大小设置为7天呢？在这种情况下，我是否也需要设置“保留”期限？

我执行KStream x KTable连接的代码如下。请注意，我已经定义了类建议和单击及其相应的serde。单击消息只是普通的字符串（url）。此URL字符串与推荐对象连接，以创建一个单击对象，该对象被发送到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();
  }

只要在joiner（上面的程序）运行后生成了建议和单击，这就可以正常工作。但是，如果在运行joiner之前为其生成建议的单击到达，我看不到任何连接发生。我该如何解决这个问题？

如果解决方案是使用KStream x KStreamjoin，请帮助我了解应选择的窗口大小和保留期。




Answer 1

你的总体观察是正确的。从概念上讲，两种方法都可以得到正确的结果。如果使用流表联接，则有两个缺点（不过在未来的Kafka版本中可能会重新讨论和改进）

• 您已经提到，如果在相应的推荐之前处理了click-get，那么（内部）连接将失败。但是，正如您所知，会有建议，您可以使用左连接而不是内部连接，检查连接结果，并在建议为null（即，您得到重试逻辑）时将单击事件写回输入主题，连续单击单个建议可能会出现问题，您可能需要在应用程序代码中对此进行说明

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

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

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

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