SparkStreaming can't connect Kafka

Question

After I have deployed my zoomeeper and Kafka clusters on Alibaba cloud server, I use my local idea to establish sparkstreamingcontext and try to connect to the Kafka cluster of the cloud server and consume data. However, the following error is reported. My code is as follows:

ERROR StreamingContext: Error starting the context, marking it as stopped org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before the position for partition first-1 could be determined

val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("StreamingWC")
val ssc = new StreamingContext(conf, Seconds(3))

ssc.checkpoint("cpp")

val kafkaPara = Map[String, String](
  ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092",
  ConsumerConfig.GROUP_ID_CONFIG -> "ryan1",
  "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
  "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)

val kafkaDS = KafkaUtils.createDirectStream[String, String](
  ssc,
  LocationStrategies.PreferConsistent,
  ConsumerStrategies.Subscribe[String, String](Set("first"), kafkaPara)
)

kafkaDS.map(_.value()).print()

ssc.start()
ssc.awaitTermination()

SparkStreaming can't connect Kafka

0 Answers