使用Spark流处理Kafka消息时的挑战

Question

1. 如果每个Kafka消息属于一个特定的会话，如何管理会话关联，以便同一个Spark执行器看到链接到一个会话的所有消息？
2. 如何确保属于会话的消息被Spark executor按照在Kafka中报告的顺序处理？我们能以某种方式实现这一点而不对线程计数施加限制并导致处理开销（如按消息时间戳排序）吗？
3. 何时检查会话状态？在执行器节点崩溃的情况下，如何从最后一个检查点恢复状态？在驱动程序节点崩溃的情况下，如何从最后一个检查点恢复状态？
4. 如果节点（执行器/驱动程序）在检查其状态之前崩溃，如何恢复状态？如果Spark通过重放消息重新创建状态RDD，那么它从哪里开始重放Kafka消息：wards上的最后一个检查点，还是处理重新创建分区所需的所有消息？Spark streaming是否可以在多个Spark streaming批处理中恢复状态，或者只对当前批处理恢复状态？例如，如果在上一批处理期间没有执行检查点，是否可以恢复状态？

Answer 1

如果每个Kafka消息属于一个特定的会话，如何管理会话亲和力，以便同一个Spark执行器看到链接到一个会话的所有消息？

Kafka将主题划分为分区，每个分区一次只能由一个使用者读取，因此您需要确保属于一个会话的所有消息都进入同一个分区。分区分配是通过分配给每个消息的密钥来控制的，因此实现这一点的最简单方法可能是在发送数据时使用会话id作为密钥。这样，同一个使用者将获得一个会话的所有消息。但是有一个警告：当一个消费者加入或离开ConsumerGroup时，Kafka将重新平衡分区分配给消费者。如果这种情况发生在会话中期，那么它可能（也将）发生，即在重新平衡之后，该会话的一半消息将传递给一个消费者，另一半将传递给另一个消费者。为了避免这种情况，您需要手动订阅代码中的特定分区，以便每个处理器都有其特定的分区集，并且不更改这些分区。查看ConsumerStrategies.SparkKafka组件代码中的指定代码。

如何确保属于会话的消息被Spark executor按照在Kafka中报告的顺序处理？我们是否可以在不对线程计数施加限制和不增加处理开销的情况下实现这一点（比如按消息时间戳排序）？

我建议阅读Spark Streaming+Kafka集成指南中的偏移存储部分，它应该已经回答了很多问题。

简短的版本是，您可以将上次读取的偏移量保存到Kafka中，并且无论何时检查执行器都应该这样做。这样，每当一个新的执行者开始处理时，不管它是否是从检查点恢复的，它都知道在Kafka中从哪里读取。

如果节点（执行器/驱动程序）在检查其状态之前崩溃，如何恢复状态？如果Spark通过重放消息重新创建状态RDD，那么它从哪里开始重放Kafka消息：wards上的最后一个检查点，还是处理重新创建分区所需的所有消息？Spark streaming可以/是否可以在多个Spark streaming批中恢复状态，或者只对当前批恢复状态，即如果在最后一个批中没有执行检查点，是否可以恢复状态？