Spark 批处理kafka指定消费条数

在 Spark 中批处理 Kafka，可以使用 `spark.streaming.kafka.maxRatePerPartition` 参数来控制每个分区每秒钟处理的最大消息数。该参数的默认值为 -1，表示不限制处理速率，可以根据实际情况进行配置。

例如，如果想要每秒钟处理每个分区 100 条消息，可以将该参数设置为 100，如下所示：

val kafkaParams = Map(
  "bootstrap.servers" -> "localhost:9092",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "group.id" -> "my-group"
)

val topics = Array("my-topic")

val stream = KafkaUtils.createDirectStream[String, String](
  streamingContext,
  PreferConsistent,
  Subscribe[String, String](topics, kafkaParams)
)

stream
  .map(record => (record.key, record.value))
  .foreachRDD { rdd =>
    rdd.foreachPartition { partitionOfRecords =>
      // Process messages in this partition
    }
  }

stream.context.sparkContext.getConf.set("spark.streaming.kafka.maxRatePerPartition", "100")

注意，该参数的限制是针对每个分区的处理速率，而非整个 Kafka 主题的处理速率。如果需要限制整个主题的处理速率，可以通过调整 Spark Streaming 的批处理时间间隔来实现。例如，将批处理时间间隔设置为 5 秒，每个分区每秒处理 100 条消息，则整个主题每秒处理 500 条消息。