Я пытаюсь подключить тему Kafak, в которой используется Avro, к Spark Stream.Я пытаюсь использовать приведенный ниже код:
schema_registry_client = CachedSchemaRegistryClient(url='schema_registry_url')
serializer = MessageSerializer(schema_registry_client)
def decoder(s):
decoded_message = serializer .decode_message(s)
return decoded_message
kvs = KafkaUtils.createDirectStream(ssc, topic, {'bootstrap.servers': brokers}, valueDecoder=decoder)
Однако я получил нижеуказанную ошибку при запуске:
У объекта 'Kafka DStream' нет атрибута '_get_object_id'
Как мы можем преодолеть это?