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

7d7tgy0s 于 2021-06-07 发布在 Kafka

关注(0)|答案(4)|浏览(360)

我有一个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（…）”
如何反序列化值？

avro scala apache-kafka spark-streaming apache-spark-2.0

来源：https://stackoverflow.com/questions/40705926/reading-avro-messages-from-kafka-with-spark-2-0-2-structured-streaming

4条答案

按热度按时间

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]
  }

}

赞(0）回复(0）举报 2021-06-07

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)
                })

赞(0）回复(0）举报 2021-06-07

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对象。
希望这对别人有帮助

赞(0）回复(0）举报 2021-06-07

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到更新的行时发生编码器错误

赞(0）回复(0）举报 2021-06-07

我来回答

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

4条答案

相关问题

热门标签

最新问答