scala—在spark streaming中反序列化来自kafka的avro格式的数据会得到空字符串,长字符串为0

dw1jzc5e  于 2021-06-07  发布在  Kafka
关注(0)|答案(1)|浏览(274)

我正在努力反序列化的avro序列化数据来自Kafka在Spark流。
这是我正在运行的文件spark submit:

package com.example.mymessage

import org.apache.avro.Schema
import org.apache.avro.generic.{GenericDatumReader, GenericRecord}
import org.apache.avro.io.DecoderFactory
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{Logging, SparkConf}
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._

object MyMessageCount extends Logging {
  def main(args: Array[String]) {
    if (args.length < 4) {
      System.err.println("Usage: MyMessageCount <zkQuorum> <group> <topics> <numThreads>")
      System.exit(1)
    }

    val log4jInitialized = Logger.getRootLogger.getAllAppenders.hasMoreElements
    if (!log4jInitialized) {
      logInfo("Setting log level to [WARN]." +
        " To override add a custom log4j.properties to the classpath.")
      Logger.getRootLogger.setLevel(Level.WARN)
    }

    val Array(zkQuorum, group, topics, numThreads) = args
    val sparkConf = new SparkConf().setMaster("local[4]").setAppName("MyMessageCount")
    val ssc = new StreamingContext(sparkConf, Seconds(2))
    ssc.checkpoint("checkpoint")

    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
    val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)

    lines.foreachRDD(rdd => {
      rdd.foreach(avroRecord => {
        val schemaString = "{\"type\":\"record\",\"name\":\"myrecord\",\"fields\":[{\"name\":\"string\",\"type\":\"string\"},{\"name\":\"long\",\"type\":\"long\"}]}"
        val parser = new Schema.Parser()
        val schema = parser.parse(schemaString)
        val reader = new GenericDatumReader[GenericRecord](schema)

        val decoder = DecoderFactory.get.binaryDecoder(avroRecord.toCharArray.map(_.toByte), null)
        val record: GenericRecord = reader.read(null, decoder)

        System.out.println(avroRecord + "," + record.toString 
          + ", string= " + record.get("string")
          + ", long=" + record.get("long"))
      })
    })

    ssc.start()
    ssc.awaitTermination()
  }
}

我一直在使用confluent平台在本地发送数据。
如果我发送:

{"string":"test","long":30}

然后上述代码输出:

test<,{"string": "", "long": 0}, string= , long=0

这向我表明数据正在通过,但出于某种原因,字符串和长值显示为类似于默认值的值。如何访问进入的真正的“string”和“long”值 avroRecord 来自Kafka?

laik7k3q

laik7k3q1#

使用confluent的kafkaavrodecoder和一个直接流就可以做到这一点。

import io.confluent.kafka.serializers.KafkaAvroDecoder

...

val kafkaParams = Map[String, String]("metadata.broker.list" -> zkQuorum,
  "schema.registry.url" -> schemaRegistry,
  "auto.offset.reset" -> "smallest")
val topicSet = Set(topics)
val messages = KafkaUtils.createDirectStream[Object, Object, KafkaAvroDecoder, KafkaAvroDecoder](ssc, kafkaParams, topicSet).map(_._2)

val lines = messages.foreachRDD(rdd => {
  rdd.foreach({ avroRecord =>
    println(avroRecord)
  })
})

我发现一个单独的问题,我只能导入版本1,而不能导入更新的版本。

libraryDependencies ++= Seq(
  "io.confluent" % "kafka-avro-serializer" % "1.0",
  ...
)

resolvers ++= Seq(
  Resolver.sonatypeRepo("public"),
  Resolver.url("confluent", url("http://packages.confluent.io/maven/"))
)

更新以下代码以获得最新版本的kafka-avro序列化程序。

libraryDependencies ++= Seq(
  "io.confluent" % "kafka-avro-serializer" % "3.0.0",
  ...
)

resolvers ++= Seq(
  Resolver.sonatypeRepo("public"),
  "Confluent Maven Repo" at "http://packages.confluent.io/maven/"
)

相关问题