使用spark 2.0.2(结构化流媒体)读取来自kafka的avro消息

7d7tgy0s  于 2021-06-07  发布在  Kafka
关注(0)|答案(4)|浏览(361)

我有一个spark2.0应用程序,它使用spark-streaming-kafka-0-10_.11从kafka读取消息。
结构化流看起来非常酷,所以我想尝试迁移代码,但我不知道如何使用它。
在常规流中,我使用kafkautils创建dstrean,在我传递的参数中是值反序列化器。
在结构化流媒体中,doc说我应该使用dataframe函数反序列化,但我不知道这到底意味着什么。
我看了一些例子,比如这个例子,但是我在Kafka中的avro对象很复杂,不能像例子中的字符串那样简单地转换。。
到目前为止,我尝试了这种代码(我在这里看到的是另一个问题):

import spark.implicits._

  val ds1 = spark.readStream.format("kafka").
    option("kafka.bootstrap.servers","localhost:9092").
    option("subscribe","RED-test-tal4").load()

  ds1.printSchema()
  ds1.select("value").printSchema()
  val ds2 = ds1.select($"value".cast(getDfSchemaFromAvroSchema(Obj.getClassSchema))).show()  
  val query = ds2.writeStream
    .outputMode("append")
    .format("console")
    .start()

我得到“数据类型不匹配:无法将binarytype转换为structtype(structfield(…)”
如何反序列化值?

p1tboqfb

p1tboqfb1#

使用以下步骤:
定义Kafka信息。
定义一个consumer实用程序,它返回avroObject的数据集。
定义逻辑代码。
Kafka信息:

case class KafkaMessage(key: String, value: Array[Byte],
                                    topic: String, partition: String, offset: Long, timestamp: Timestamp)

Kafka消费者:

import java.util.Collections

import com.typesafe.config.{Config, ConfigFactory}
import io.confluent.kafka.serializers.KafkaAvroDeserializer
import org.apache.avro.Schema
import org.apache.avro.generic.GenericRecord
import org.apache.spark.sql.SparkSession

import scala.reflect.runtime.universe._

object KafkaAvroConsumer {

  private val conf: Config = ConfigFactory.load().getConfig("kafka.consumer")
  val valueDeserializer = new KafkaAvroDeserializer()
  valueDeserializer.configure(Collections.singletonMap("schema.registry.url",
    conf.getString("schema.registry.url")), false)

  def transform[T <: GenericRecord : TypeTag](msg: KafkaMessage, schemaStr: String) = {
    val schema = new Schema.Parser().parse(schemaStr)
    Utils.convert[T](schema)(valueDeserializer.deserialize(msg.topic, msg.value))
  }

  def createDataStream[T <: GenericRecord with Product with Serializable : TypeTag]
  (schemaStr: String)
  (subscribeType: String, topics: String, appName: String, startingOffsets: String = "latest") = {

    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName(appName)
      .getOrCreate()

    import spark.implicits._

    // Create DataSet representing the stream of KafkaMessage from kafka
    val ds = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", conf.getString("bootstrap.servers"))
      .option(subscribeType, topics)
      .option("startingOffsets", "earliest")
      .load()
      .as[KafkaMessage]
      .map(msg => KafkaAvroConsumer.transform[T](msg, schemaStr)) // Transform it Avro object.

    ds
  }

}

更新
实用程序:

import org.apache.avro.Schema
import org.apache.avro.file.DataFileReader
import org.apache.avro.generic.{GenericDatumReader, GenericRecord}
import org.apache.avro.specific.SpecificData

import scala.reflect.runtime.universe._

object Utils {

  def convert[T <: GenericRecord: TypeTag](targetSchema: Schema)(record: AnyRef): T = {
      SpecificData.get.deepCopy(targetSchema, record).asInstanceOf[T]
  }

}
mmvthczy

mmvthczy2#

如上所述,从spark2.1.0开始,批处理读取器支持avro,但sparksession.readstream()不支持。下面是我如何让它在scala中工作的其他React的基础上。为了简洁起见,我简化了这个模式。

package com.sevone.sparkscala.mypackage

import org.apache.spark.sql._
import org.apache.avro.io.DecoderFactory
import org.apache.avro.Schema
import org.apache.avro.generic.{GenericDatumReader, GenericRecord}

object MyMain {

    // Create avro schema and reader
    case class KafkaMessage (
        deviceId: Int,
        deviceName: String
    )
    val schemaString = """{
        "fields": [
            { "name":  "deviceId",      "type": "int"},
            { "name":  "deviceName",    "type": "string"},
        ],
        "name": "kafkamsg",
        "type": "record"
    }""""
    val messageSchema = new Schema.Parser().parse(schemaString)
    val reader = new GenericDatumReader[GenericRecord](messageSchema)
    // Factory to deserialize binary avro data
    val avroDecoderFactory = DecoderFactory.get()
    // Register implicit encoder for map operation
    implicit val encoder: Encoder[GenericRecord] = org.apache.spark.sql.Encoders.kryo[GenericRecord]

    def main(args: Array[String]) {

        val KafkaBroker =  args(0);
        val InTopic = args(1);
        val OutTopic = args(2);

        // Get Spark session
        val session = SparkSession
                .builder
                .master("local[*]")
                .appName("myapp")
                .getOrCreate()

        // Load streaming data
        import session.implicits._
        val data = session
                .readStream
                .format("kafka")
                .option("kafka.bootstrap.servers", KafkaBroker)
                .option("subscribe", InTopic)
                .load()
                .select($"value".as[Array[Byte]])
                .map(d => {
                    val rec = reader.read(null, avroDecoderFactory.binaryDecoder(d, null))
                    val deviceId = rec.get("deviceId").asInstanceOf[Int]
                    val deviceName = rec.get("deviceName").asInstanceOf[org.apache.avro.util.Utf8].toString
                    new KafkaMessage(deviceId, deviceName)
                })
uttx8gqw

uttx8gqw3#

所以实际上我公司有人帮我解决了这个问题,所以我会把它贴在这里给未来的读者。。
基本上,我错过了米古诺建议的解码部分:

def decodeMessages(iter: Iterator[KafkaMessage], schemaRegistryUrl: String) : Iterator[<YourObject>] = {
val decoder = AvroTo<YourObject>Decoder.getDecoder(schemaRegistryUrl)
iter.map(message => {
  val record = decoder.fromBytes(message.value).asInstanceOf[GenericData.Record]
  val field1 = record.get("field1Name").asInstanceOf[GenericData.Record]
  val field2 = record.get("field1Name").asInstanceOf[GenericData.String]
        ...
  //create an object with the fields extracted from genericRecord
  })
}

现在,你可以阅读Kafka的信息,并像这样解码它们:

val ds = spark
  .readStream
  .format(config.getString(ConfigUtil.inputFormat))
  .option("kafka.bootstrap.servers", config.getString(ConfigUtil.kafkaBootstrapServers))
  .option("subscribe", config.getString(ConfigUtil.subscribeTopic))
  .load()
  .as[KafkaMessage]

val decodedDs  = ds.mapPartitions(decodeMessages(_, schemaRegistryUrl))
  • KafkaMessage 只是一个case类,它包含从kafka读取时得到的泛型对象

(key,value,topic,partition,offset,timestamp) AvroTo<YourObject>Decoder 是一个类,它将在给定模式注册表url的情况下解码对象。
例如使用confluent的 KafkaAvroDeserializer 和架构注册表。

val kafkaProps = Map("schema.registry.url" -> schemaRegistryUrl)
val client = new CachedSchemaRegistryClient(schemaRegistryUrl, 20)

// If you have Avro encoded keys
val keyDeserializer = new KafkaAvroDeserializer(client)
keyDeserializer.configure(kafkaProps.asJava, true) //isKey = true

// Avro encoded values
valueDeserializer = new KafkaAvroDeserializer(client)
valueDeserializer.configure(kafkaProps.asJava, false) //isKey = false

从这些,打电话 .deserialize(topicName, bytes).asInstanceOf[GenericRecord] 获取avro对象。
希望这对别人有帮助

kuarbcqp

kuarbcqp4#

我还不太熟悉spark的序列化是如何与新的/实验性的结构化流媒体结合使用的,但是下面的方法确实有效——尽管我不确定这是不是最好的方法(我觉得这种方法有点尴尬)。
我将尝试在一个自定义数据类型的示例中回答您的问题(这里是 Foo 案例类),而不是具体的avro,但我希望它会帮助你无论如何。其思想是使用kryo序列化来序列化/反序列化您的自定义类型,请参阅spark文档中的tuning:data serialization。
注意:spark支持通过内置(隐式)编码器对case类进行开箱即用的序列化,您可以通过 import spark.implicits._ . 但是为了这个例子,让我们忽略这个功能。
假设您定义了以下内容 Foo case类作为自定义类型(tl;提示:为了防止遇到奇怪的投诉/错误,你应该把代码放到一个单独的 Foo.scala 文件):

// This could also be your auto-generated Avro class/type
case class Foo(s: String)

现在您有了以下结构化流式代码来从kafka读取数据,其中输入主题包含kafka消息,其消息值是二进制编码的 String ,你的目标是创造 Foo 基于这些消息值的示例(例如,类似于将二进制数据反序列化为avro类示例的方式):

val messages: DataFrame = spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
    .option("subscribe", "my-input-topic")
    .load()

现在,我们正在将这些值反序列化到自定义的示例中 Foo 类型,首先需要为其定义一个隐式 Encoder[Foo] :

implicit val myFooEncoder: Encoder[Foo] = org.apache.spark.sql.Encoders.kryo[Foo]
val foos: Dataset[Foo] = messages.map(row => Foo(new String(row.getAs[Array[Byte]]("value")))

回到你的avro问题,你需要做的是:
创建一个适当的 Encoder 为了你的需要。
替换 Foo(new String(row.getAs[Array[Byte]]("value")) 使用代码将二进制编码的avro数据反序列化为avro pojo,即从消息值中取出二进制编码的avro数据的代码( row.getAs[Array[Byte]]("value") )返回,比如说,一个avro GenericRecord 或者别的什么 SpecificCustomAvroObject 你已经在别处定义了。
如果其他人知道一个更简洁/更好的/。。。回答塔尔问题的方式,我洗耳恭听
另请参见:
如何在数据集中存储自定义对象?
尝试将Dataframe行Map到更新的行时发生编码器错误

相关问题