Всякий раз, когда я перезапускаю систему, он показывает, что кодек utf-8 не может декодировать байт 0x98 в kafka, чтобы зажечь

Это мой следующий код для получения данных от kafka для запуска потоковой передачи Сначала он работал, но когда я перезапустил систему, он снова показывает следующую ошибку:

UnicodeDecodeError: кодек utf-8 не может декодировать байт 0x98 в позиции 5: недопустимый начальный байт

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from pyspark.streaming.kafka import KafkaUtils
from confluent_kafka.avro.cached_schema_registry_client import CachedSchemaRegistryClient
from confluent_kafka.avro.serializer.message_serializer import MessageSerializer

schema_registry_client = CachedSchemaRegistryClient(url='http://0.0.0.0:8081')
serializer = MessageSerializer(schema_registry_client)
sc = SparkContext()
sc.setLogLevel("ERROR")
ssc = StreamingContext(sc, 2)



# def decoder(s):
#     decoded_message = serializer.decode_message(s)
#     return decoded_message

kvs = KafkaUtils.createDirectStream(ssc, ["demo.Appointment_Attendance.Patient"], {
                                "metadata.broker.list": "localhost:9092"}, 
                                 valueDecoder=serializer.decode_message)
keyDecoder=lambda x: x, valueDecoder=lambda x: x)
lines = kvs.map(lambda x: x[1])
lines.pprint()

ssc.start()
ssc.awaitTermination()