使用Scala Jackson库使用Spark加载和解析JSON

Question

我正在尝试加载和解析json文件（推文），但我收回了以下错误

error: not found: value mapper
               Some(mapper.readValue(record, classOf[Tweet]))

这是scala脚本

import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.DeserializationFeature

case class Tweet(tweet_id: Int, created_unixtime: Long, created_time: String, lang: String, displayname: String, time_zone: String, msg: String)

val = input.textFile("hdfs://localhost:54310/tmp/data_staging/tweets*") // tweets well loaded

// Parsing them
val result = input.flatMap(record => {
  try {
    Some(mapper.readValue(record, classOf[Tweet]))
  } catch {
    case e: Exception => None
  }
})

Answer 1

所以，问题是如何加载JSON但将其映射到案例类。

在这种情况下，只需使用Spark的内置JSON阅读器，然后转换为案例类的DataSet：

case class Tweet(tweet_id: Int, created_unixtime: Long, created_time: String, lang: String, displayname: String, time_zone: String, msg: String)

val input = spark.read.json("hdfs://localhost:54310/tmp/data_staging/tweets*").as[Tweet]

这里的假设是JSON文档中的字段映射到您的案例类。如果不是这种情况，那么您只需执行一个映射即可将Row对象转换为自定义案例类。