如何通过Spark从Kafka那里获得至少N个日志？

Question

在Spark streaming中，我会在日志到达时获取日志。但我希望在一次传递中获得至少N个日志。如何才能实现呢？

从这个答案来看，Kafka似乎有这样一种效用，但在Spark中似乎并不存在，使之成为可能。

Answer 1

没有选项允许您设置从Kafka接收的消息数的最小值。选项maxoffsetspertrigger允许您设置最大消息数。

如果您希望您的微批处理一次处理更多的消息，您可以考虑增加您的触发间隔。

还有（参考你提供的链接），这也是不可能在Kafaka本身设置的。您可以设置提取字节数的最小数量，但不能设置消息数的最小数量。

这样，您还可以使用使用者配置kafka.fetch.min.bytes。但是，在loval Kafka 2.5.0安装上用Spark 3.0.1测试这一点并没有任何影响。在添加配置kafka.fetch.max.wait.ms时，我的测试中的提取定时发生了变化，但变化的方式并不是可预测的（至少对我来说是这样）。

查看Spark的KafkaDataConsumer的源代码，与纯KafkaConsumer相比，fetch似乎不直接占任何最小/最大字节。