Lesen Avro Nachrichten von Kafka mit Spark 2.0.2 (strukturierte streaming)

Ich bin noch nicht so super vertraut, wie die Spark-Serialisierung funktioniert in Kombination mit der neuen/experimentellen Strukturierte Streaming, aber der Ansatz unten funktioniert-obwohl ich bin nicht sicher, ob es der beste Weg (IMHO der Ansatz hat eine etwas peinliche look 'n feel).

Ich werde versuchen, Ihre Frage zu beantworten am Beispiel eines benutzerdefinierten Datentyps (hier: ein Foo Fall Klasse) statt, die speziell Avro, aber ich hoffe, es hilft dir sowieso. Die Idee ist die Verwendung von Kryo-Serialisierung serialisieren/Deserialisieren Ihre benutzerdefinierte Typ finden Tuning: - Daten-Serialisierung in der Funke Dokumentation.

Hinweis: Spark unterstützt die Serialisierung von case-Klassen aus der box über eingebaute (implizite) Encoder, die Sie importieren können über import spark.implicits._. Aber wir ignorieren diese Funktionalität aus Gründen der diesem Beispiel.

Stellen Sie sich vor Sie haben folgende Foo Fall Klasse, wie Sie Ihre benutzerdefinierten Typ (TL;DR Hinweis: um zu verhindern, laufen in komischen Funken die Serialisierung von Beschwerden/Fehlern sollten Sie den code in einem separaten Foo.scala - Datei):

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

Nun haben Sie folgende Strukturierten Streaming-code zum Lesen von Daten von Kafka, wobei die Eingabe Thema enthält Kafka Nachrichten, deren Meldung Wert eine Binär codierte String, und Ihr Ziel ist es zu schaffen Foo - Instanzen basierend auf diese Nachricht Werte (also ähnlich wie du würde Deserialisieren von binären Daten in Instanzen von einem Avro-Klasse):

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

Nun sind wir Deserialisieren die Werte in Instanzen der benutzerdefinierten Foo geben, für die müssen wir zuerst definieren eine implizite 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")))

Gehen Sie zurück zu Ihrem Avro Frage, was Sie würden tun müssen, ist:

1. Erstellen Sie eine korrekte Encoder für Ihre Bedürfnisse.
2. Ersetzen Foo(new String(row.getAs[Array[Byte]]("value")) mit dem code zum Deserialisieren Ihre Binär codierte Avro Daten in Avro POJOs, d.h. dem code, der Ihre Binär codierte Avro Daten aus der Nachricht mit dem Wert (row.getAs[Array[Byte]]("value")) und zurück, sagen wir, eine Avro GenericRecord oder was auch immer SpecificCustomAvroObject Sie haben an anderer Stelle definierten.

Wenn jemand anderes weiß, der eine präziser/besser/... Antwort Tal der Frage, bin ich alle Ohren. 🙂

Siehe auch:

• Wie zum speichern benutzerdefinierter Objekte im Dataset?
• Encoder-Fehler beim anzeigen dataframe Zeile zu Zeile aktualisiert

Also tatsächlich jemand in meiner Firma das Problem gelöst für mich, Also poste ich es hier für zukünftige Leser.

im Grunde, was ich habe auf der Spitze von dem, was miguno vorgeschlagen, ist die decode-Teil:

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

jetzt können Sie Lesen Nachrichten aus kafka und decodieren Sie Sie wie so:

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 ist einfach nur Klasse mit dem generischen Objekt, das Sie erhalten, wenn das Lesen von Kafka (key,value,topic,partition,offset,timestamp)

AvroTo<YourObject>Decoder ist Klasse, dass werde Dekodieren, Ihr Objekt in einem gegebenen schema Registrierungs-url.

Beispielsweise mit Konfluent ist KafkaAvroDeserializer und schema-Registrierung.

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

Aus diesen, nennen .deserialize(topicName, bytes).asInstanceOf[GenericRecord] um eine avro-Objekt.

Hoffe, dies hilft jemand