Spark Stream - 'utf8' 编解码器无法解码字节答案

【问题标题】：Spark Stream - 'utf8' codec can't decode bytesSpark Stream - 'utf8' 编解码器无法解码字节
【发布时间】：2019-03-13 03:11:24
【问题描述】：

我对流式编程相当陌生。我们有使用 Avro 的 Kafka 流。

我想将 Kafka Stream 连接到 Spark Stream。我使用了下面的代码。

kvs = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": brokers})
lines = kvs.map(lambda x: x[1])

我收到以下错误。

return s.decode('utf-8') 解码中的文件“/usr/lib64/python2.7/encodings/utf_8.py”，第 16 行返回 codecs.utf_8_decode（输入，错误，真） UnicodeDecodeError：“utf8”编解码器无法解码位置 57-58 中的字节：无效的继续字节

我是否需要指定 Kafka 使用 Avro，是否存在上述错误？如果是我如何指定它？

【问题讨论】：

标签： python apache-spark pyspark spark-streaming

【解决方案1】：

是的，问题在于流的反序列化。您可以使用 confluent-kafka-python 库并在 valueDecoder 中指定：

from confluent_kafka.avro.cached_schema_registry_client import CachedSchemaRegistryClient`
from confluent_kafka.avro.serializer.message_serializer import MessageSerializer

kvs = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": brokers}, valueDecoder=MessageSerializer.decode_message)`

https://stackoverflow.com/a/49179186/6336337 解决方案的功劳

【讨论】：

您好 Lev G，我认为您的回答是正确的，但是在进行更改后我又遇到了一个错误。
但是我将 serializer.decode_message 更改为 MessageSerializer.decode_message。可以吗？新错误：stackoverflow.com/q/52714436/8089635
对，我修正了我的答案 - 它应该是 MessageSerializer。请采纳答案
在 Kafka 中我们使用的是 Schema Registry，我们不应该在这里指定它吗？

【解决方案2】：

是的，您应该指定它。

用java：

流的创建：

final JavaInputDStream<ConsumerRecord<String, avroType>> stream =
                KafkaUtils.createDirectStream(
                        jssc,
                        LocationStrategies.PreferConsistent(),
                        ConsumerStrategies.Subscribe(topics, kafkaParams));

在 kafka 消费者配置中：

kafkaParams.put("key.deserializer", org.apache.kafka.common.serialization.StringDeserializer.class);
        kafkaParams.put("value.deserializer", SpecificAvroDeserializer.class);

【讨论】：