如何使用spark批量加载kafka主题中的所有记录

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
    <version>3.2.0</version>
</dependency>

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

public class KafkaSparkStreamingExample {
    public static void main(String[] args) throws InterruptedException {
        // 创建Spark配置
        SparkConf conf = new SparkConf().setAppName("KafkaSparkStreamingExample").setMaster("local[*]");

        // 创建Streaming上下文
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

        // 创建Kafka消费者参数
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "kafka-server1:9092,kafka-server2:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "kafka-consumer-group");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", false);

        // 创建Kafka主题输入流
        JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream(
                jssc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(Collections.singleton("kafka-topic"), kafkaParams)
        );

        // 处理接收到的数据
        stream.foreachRDD(rdd -> {
            rdd.foreach(record -> {
                // 在这里进行数据处理，可以根据业务需求进行相应的操作
                System.out.println(record.value());
            });
        });

        // 启动Spark Streaming应用
        jssc.start();

        // 等待应用完成
        jssc.awaitTermination();
    }
}