以前的kafkautils包是“org.apache.spark.streaming.kafka”。最新的包是“org.apache.spark.streaming.kafka010”。
要设置kafkaparams和主题详细信息，请检查以下代码段，

import java.util.*;
import org.apache.spark.SparkConf;
import org.apache.spark.TaskContext;
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.api.java.*;
import org.apache.spark.streaming.kafka010.*;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import scala.Tuple2;

Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "localhost:9092,anotherhost:9092");
kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", StringDeserializer.class);
kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
kafkaParams.put("auto.offset.reset", "latest");
kafkaParams.put("enable.auto.commit", false);

Collection<String> topics = Arrays.asList("topicA", "topicB");

final JavaInputDStream<ConsumerRecord<String, String>> stream =
  KafkaUtils.createDirectStream(
    streamingContext,
    LocationStrategies.PreferConsistent(),
    ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
  );

stream.mapToPair(
  new PairFunction<ConsumerRecord<String, String>, String, String>() {
    @Override
    public Tuple2<String, String> call(ConsumerRecord<String, String> record) {
      return new Tuple2<>(record.key(), record.value());
    }
  })

如需进一步参考，请访问以下链接https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

kafkautils是apachespark流媒体的一部分，而不是apachekafka的一部分
org.apache.spark.streaming.kafka.kafkautils网站