spark streaming kafka010 consumer thow l lot of errors

classic Classic list List threaded Threaded
1 message Options
Reply | Threaded
Open this post in threaded view
|

spark streaming kafka010 consumer thow l lot of errors

yunchen liu
Hey everyone,

    spark streaming consumer kafka010 throw a lot of errors, spark program does not exit, but after running for some time many Executor loss and no error.

    spark  version: 2.4.3
    kafka version: 0.10

org.apache.kafka.common.errors.InterruptException: java.lang.InterruptedException
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:447)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:254)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:214)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:205)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:137)
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:307)
at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1138)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1103)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.poll(CachedKafkaConsumer.scala:99)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.get(CachedKafkaConsumer.scala:73)
at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:223)
at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:189)
at streaming.VehicelVsLite$$anonfun$main$3$$anonfun$1.apply(VehicelVsLite.scala:139)
at streaming.VehicelVsLite$$anonfun$main$3$$anonfun$1.apply(VehicelVsLite.scala:135)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:801)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:801)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:121)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.InterruptedException

at streaming.VehicelVsLite$$anonfun$main$3$$anonfun$1.apply(VehicelVsLite.scala:139)
at streaming.VehicelVsLite$$anonfun$main$3$$anonfun$1.apply(VehicelVsLite.scala:135)

val vsRDD = v.mapPartitions(messageIt => {
val signalBuffer = new ArrayBuffer[String]()
while(messageIt.hasNext){
try {
val record = messageIt.next()
val message = record.value()
val topic = record.topic()

val uncompressed = FileUtils.uncompressGz(message)
val signal = new String(uncompressed, "UTF-8")
val rq = JSON.parseObject(signal)
val vehType=rq.getString("VehicleModel")
//TODO 只解析as28lite
if(vehType!=null){
if(vehType.equals("100021")){
val oem = "saic"
val ret = VsParserForSaic.process(oem,rq)
signalBuffer +=ret
}
}
} catch {
case inte: InterruptedException = >
log.error("Publisher thread interrupted. Exception: {}.", record)
case ex: Exception =>
ex.printStackTrace()
log.info(ex.getMessage)
}
}
signalBuffer.iterator
})

any solution for this issue ?

Many Thanks.
YunKillEroe