Невозможно получить доступ к десериализованным вложенным элементам записи avro generi c в scala - PullRequest
Новая
       39

Невозможно получить доступ к десериализованным вложенным элементам записи avro generi c в scala

0 голосов
/ 26 марта 2020

Я использую структурированную потоковую передачу (Spark 2.4.0) для чтения авро-сообщений через kafka и использую Confluent schema-Registry для получения / чтения схемы

Я не могу получить доступ к глубоко вложенным полям.

Схема выглядит следующим образом в сжатом формате avs c: 

{"type":"record","name":"KafkaMessage","namespace":"avro.pojo","fields":[{"name":"context","type":["null",{"type":"record","name":"Context","fields":[{"name":"businessInteractionId","type":["null","string"]},{"name":"referenceNumber","type":["null","string"]},{"name":"serviceName","type":["null","string"]},{"name":"status","type":["null","string"]},{"name":"sourceSystems","type":["null",{"type":"array","items":{"type":"record","name":"SourceSystem","fields":[{"name":"orderId","type":["null","string"]},{"name":"revisionNumber","type":["null","string"]},{"name":"systemId","type":["null","string"]}]}}]},{"name":"sysDate","type":["null","string"]}]}]}]}

как проанализировано в искре 

      context
     |-- businessInteractionId: string (nullable = true)
     |-- referenceNumber: string (nullable = true)
     |-- serviceName: string (nullable = true)
     |-- sourceSystems: array (nullable = true)
     |    |-- element: struct (containsNull = true)
     |    |    |-- orderId: string (nullable = true)
     |    |    |-- revisionNumber: string (nullable = true)
     |    |    |-- systemId: string (nullable = true)
     |-- status: string (nullable = true)
     |-- sysDate: string (nullable = true)

Мой подход: приведите возвращенный объект как GenericRecord и массив как GenericData.Array [GenericRecord] Ссылка

Код 

val client = new CachedSchemaRegistryClient(schemaRegUrl, 100)
val brdDeser = spark.sparkContext.broadcast(new KafkaAvroDeserializer(client).asInstanceOf[Deserializer[GenericRecord]])

    val results = df.select(col("value").as[Array[Byte]]).map {
      rawBytes: Array[Byte] =>
        //read the raw bytes from spark and then use the confluent deserializer to get the record back
        val deser = brdDeser.value
        val decoded = deser.deserialize(topics, rawBytes)
        val context_GR =
          decoded.get("context").asInstanceOf[GenericRecord]
        val c_businessInteractionId =
          context_GR.get("businessInteractionId").toString  //this works
        val c1_sourceSystems =
          context_GR
            .get("sourceSystems")
            .asInstanceOf[GenericData.Array[GenericRecord]]
        val c_orderId = c1_sourceSystems.get(0).get("orderId").toString   //NullPointerException
        val c_revisionNumber = c1_sourceSystems.get(0).get("revisionNumber").toString
        val c_systemId = c1_sourceSystems.get(0).get("systemId").toString
new CaseMessage(
          c_businessInteractionId, c_orderId, c_revisionNumber, c_systemId )
    }

case class CaseMessage(c_businessInteractionId: String,
                         c_orderId: String,
                         c_revisionNumber: String,
                         c_systemId: String,)

Каждый раз, когда я получаю исключение java .lang.NullPointerException, когда оно пытается оценить c_orderId

1 Ответ

0 голосов
/ 01 апреля 2020

Это была проблема с данными. Я смог решить эту проблему, выполнив проверку нулевого значения 

val c_orderId = if (c1_sourceSystems.get(0).get("orderId") != null) {
          c1_sourceSystems.get(0).get("orderId").toString
...