我从 Kafka 获取数据,然后使用默认解码器反序列化
Array[Byte]
,之后我的 RDD 元素看起来像 (null,[B@406fa9b2)
、(null,[B@21a9fe0)
,但我想要具有模式的原始数据,那么我该如何实现这一点?
我以 Avro 格式序列化消息。
您必须使用适当的反序列化器对字节进行解码,例如字符串或自定义对象。
如果你不进行解码,你会得到
[B@406fa9b2
,这只是 Java 中字节数组的文本表示。
Kafka 对消息的内容一无所知,因此它将字节数组从生产者传递到消费者。
在 Spark Streaming 中,您必须对键和值使用序列化器(引用 KafkaWordCount 示例):
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer")
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer")
使用上述序列化器,您将获得
DStream[String]
,因此您可以使用 RDD[String]
。
但是,如果您想直接将字节数组反序列化为自定义类,则必须编写一个自定义 Serializer (这是 Kafka 特定的,与 Spark 无关)。
我建议使用具有固定模式的 JSON 或 Avro(使用Kafka、Spark 和 Avro - 第 3 部分,生成和使用 Avro 消息中描述的解决方案)。
在结构化流媒体中,管道可能如下所示:
val fromKafka = spark.
readStream.
format("kafka").
option("subscribe", "topic1").
option("kafka.bootstrap.servers", "localhost:9092").
load.
select('value cast "string") // <-- conversion here
以下答案是他答案的延伸。
如果您使用 Spark 的结构化流来消费数据,那么您可以执行以下操作:
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
// let us say your spark session is defined with 'spark'
val readDF = spark.readStream
.format("kafka")
.option("subscribe", "topic1")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("groupIdPrefix","whatever-group-id")
.option("startingOffsets","latest")
.load
.select(col("value").cast(StringType))
您可以在生产者端编写任何您想要的数据类型,我以 String 为例。