有没有可能发送一个java对象作为kafka主题中的值，以及如何在spark中使用它？
我目前正在做apachespark教程，想知道是否可以发送字符串以外的内容。本教程有以下示例
producer.send(new ProducerRecord<String, String>(topic, something_string));有可能这样做吗？
Car car = new Car(brand, year, color);
producer.send(new ProducerRecord<String, Car>(topic, car));在spark的后期我该如何消耗它呢？
目前，我正在这样做：
String car = brand + "," + year + "," + color;
producer.send(new ProducerRecord<String, String>(topic, car));我把所有的东西都放在一个用逗号分隔的字符串里。
问题二：目前我是这样消费的。
Dataset<String> words = df
.selectExpr("CAST (value AS STRING)")
.as(Encoders.STRING());我得到字符串的地方： "brand,year,color" 我该如何将它拆分并放在不同的列中？
1条答案
按热度按时间bxfogqkk1#
你的帖子实际上有两个问题，你可以把它们分成不同的帖子。关于第一个问题，请参阅本帖；核心概念是必须编写自定义序列化程序。
对于第二种情况，这个概念在原则上仍然是相同的，但是这次您必须在spark端编写一个定制的反序列化器（解码器）。参考spark文档，它演示了如何从kafka创建流。但是，请不要使用“kafkautil”类，请参阅javadoc。它具有使用kafka解码器类创建流的方法。