如何使用scala在kafka消费者中实现反序列化？

Question

我的 kafka 消费者代码中有以下行。

val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicpMap).map(_._2) 

如何将此流“行”反序列化为原始对象？通过将类扩展为可序列化，在 kafka 生产者中实现了可序列化。我正在使用 scala 在 spark 中实现这一点。

Answer 1

您需要实现自定义Decoder，并将预期的类型信息与解码器一起提供给createStream 函数。

KafkaUtils.createStream[KeyType, ValueType, KeyDecoder, ValueDecoder] (...)

例如，如果您使用String 作为键和CustomContainer 作为值，您的流创建将如下所示：

val stream = KafkaUtils.createStream[String, CustomContainer, StringDecoder, CustomContainerDecoder](...)  

鉴于您将消息编码为new KeyedMessage[String,String]，因此正确的解码器是这样的字符串解码器：

KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](topic,...)

这将为您提供DStream[String,String] 作为您处理的基础。

如果你想发送/接收特定的对象类型，你需要为它实现一个 Kafka Encoder 和 Decoder。 幸运的是，PcapPacket 已经实现了您需要的方法：

• PcapPacket -> byte[]: public int transferStateAndDataTo(byte[] buffer)

• byte[] -> PcapPacket: public PcapPacket(byte[] buffer)

剩下的就是实现 Kafka 所需的 Encoder/Decoder 接口的样板代码。