将 spark 数据帧 Row 转换为 Avro 并发布到 kafka答案

【问题标题】：Convert a spark dataframe Row to Avro and publish to kakfa将 spark 数据帧 Row 转换为 Avro 并发布到 kafka
【发布时间】：2018-09-01 09:02:27
【问题描述】：

我有一个具有以下架构的 spark 数据帧，并尝试使用 Avro 将此数据帧流式传输到 Kafka

```root
 |-- clientTag: struct (nullable = true)
 |    |-- key: string (nullable = true)
 |-- contactPoint: struct (nullable = true)
 |    |-- email: string (nullable = true)
 |    |-- type: string (nullable = true)
 |-- performCheck: string (nullable = true)```

样本记录：{"performCheck" : "N", "clientTag" :{"key":"value"}, "contactPoint": {"email":"abc@gmail.com", "type":"EML"}}

Avro 架构：

{ "name":"Message", "namespace":"kafka.sample.avro", "type":"record", "fields":[ {"type":"string", "name":"id"}, {"type":"string", "name":"email"} {"type":"string", "name":"type"} ] }

我有几个问题。

将org.apache.spark.sql.Row 转换为Avro 消息的最佳方法是什么，因为我想从每一行的数据框中提取email 和type 并使用这些值来构造Avro 消息？
最终，所有的 Avro 消息都将发送到 Kafka。那么，如果生产过程中出现错误，如何将所有未能生产的 Row 收集到 Kafka 并返回数据帧？

感谢您的帮助

【问题讨论】：

标签： apache-spark dataframe apache-kafka spark-dataframe spark-streaming

【解决方案1】：

你可以试试这个。

Q#1：您可以使用点符号提取数据框的子元素：

  val dfJSON = spark.read.json("/json/path/sample_avro_data_as_json.json") //can read from schema registry
    .withColumn("id", $"clientTag.key")
    .withColumn("email", $"contactPoint.email")
    .withColumn("type", $"contactPoint.type")

然后您可以直接使用这些列，同时将值分配给您序列化并发送到 Kafka 的 Avro 记录。

Q#2：您可以像这样跟踪成功和失败。这不是完全有效的代码，但可以给你一个想法。

  dfJSON.foreachPartition( currentPartition => {

    var producer = new KafkaProducer[String, Array[Byte]](props)
    var schema: Schema = ...//Get schema from schema registry or avsc file
    val schemaRegProps = Map("schema.registry.url" -> schemaRegistryUrl)
    val client = new CachedSchemaRegistryClient(schemaRegistryUrl, Int.MaxValue)
    valueSerializer = new KafkaAvroSerializer(client)
    valueSerializer.configure(schemaRegProps, false)

    val failedRecDF = currentPartition.map(rec =>{
      try {

        var avroRecord: GenericRecord = new GenericData.Record(schema)
        avroRecord.put("id", rec.getAs[String]("id"))
        avroRecord.put("email", rec.getAs[String]("email"))
        avroRecord.put("type", rec.getAs[String]("type"))

        // Serialize record in Producer record & send to Kafka

        producer.send(new ProducerRecord[String, Array[Byte]](kafkaTopic, rec.getAs[String]("id").toString(), valueSerializer.serialize(kafkaTopic, avroRecord).toArray))
        (rec.getAs[String]("id"), rec.getAs[String]("email"), rec.getAs[String]("type"), "Success")
      }catch{
        case e: Exception => println("*** Exception *** ")
          e.printStackTrace()

          (rec.getAs[String]("id"), rec.getAs[String]("email"), rec.getAs[String]("type"), "Failed")
      }

    })//.toDF("id", "email", "type", "sent_status")

    failedRecDF.foreach(println)
    //You can retry or log them
  })

回应是：

(111,abc@gmail.com,EML,Success)

你可以用它做任何你想做的事情。

【讨论】：