Pyspark-Kafka java.lang.ClassCastException

Question

We are getting a CAST EXCEPTION when trying to pass the offsets from Pyspark code to Kafka DStream.

py4j.protocol.Py4JJavaError: An error occurred while calling o2938.createDirectStreamWithMessageHandler.
: java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.Long

We have upgraded from Python 2.7 to 3.6 and for the upgrade we had to change the long() to int() under the Offset function, on doing the change we are getting the CAST exception!!!

kafka_stream = KafkaUtils.createDirectStream(ssc=spark_streaming_ctx,
                                           topics="Test",
                                           kafkaParams={
                                               "BROKER": "BROKER_LIST"),
                                               "auto.offset.reset": "smallest"},
                                           fromOffsets=self.Offset(spark, "TOPIC"),
                                           messageHandler="messageHandler")

def Offset(in_spark, in_topic_list):
    offsets_map = {}
    offsets_map[Topic] = int(offset[1])  # convert long to int for Python3X
    return offsets_map

We also tried use the future library https://pypi.org/project/past/, yet we didn't have any luck.

Python : 3.6 Pyspark : 2.4 Kafka : 2.1.0

did you try with float(offset[1]) ?

Flo
– Flo

2019-07-17 15:15:20 +00:00
Commented Jul 17, 2019 at 15:15 — Flo
– Flo, Commented Jul 17, 2019 at 15:15
yes @flo tried that too, no luck with the same

DataWrangler
– DataWrangler

2019-07-17 15:16:00 +00:00
Commented Jul 17, 2019 at 15:16 — DataWrangler
– DataWrangler, Commented Jul 17, 2019 at 15:16

Muthu M · Accepted Answer · 2019-07-27 10:45:40Z

1

Please check your spark submit code in case if you are using 0_10 version streaming jar file You will get above error. 0_10 version will not support python. Instead of 0_10 you can use 0_8 Kafka version.

Kafka 0_8 Version will accept the integer offset value.

Could you please refer below document.

https://spark.apache.org/docs/2.2.0/streaming-kafka-integration.html

edited Jul 27, 2019 at 10:45

answered Jul 27, 2019 at 8:50

Muthu M

262 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

DataWrangler Over a year ago

Thanks that solved the issue passing the corresponding jar made the trick.

Collectives™ on Stack Overflow

Pyspark-Kafka java.lang.ClassCastException

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related