Spark Dataframe - 编码器答案

【问题标题】：Spark Dataframe - EncoderSpark Dataframe - 编码器
【发布时间】：2020-12-07 05:08:41
【问题描述】：

我是 Scala 和 Spark 的新手。

我正在尝试使用编码器从 Spark 读取文件，然后转换为 java/scala 对象。

使用 as 读取应用架构和编码的文件的第一步工作正常。

然后我使用该数据集/数据框进行简单的地图操作，但如果我尝试在生成的数据集/数据框上打印架构，它不会打印任何列。

另外，当我第一次读取文件时，我没有在 Person 类中映射年龄字段，只是为了在 map 函数中计算它来尝试 - 但我没有看到年龄没有使用映射到数据框完全是人。

Person.txt 中的数据：

firstName,lastName,dob
ABC, XYZ, 01/01/2019
CDE, FGH, 01/02/2020

下面是代码：

object EncoderExample extends App {
  val sparkSession = SparkSession.builder().appName("EncoderExample").master("local").getOrCreate();

  case class Person(firstName: String, lastName: String, dob: String,var age: Int = 10)
  implicit val encoder = Encoders.bean[Person](classOf[Person])
  val personDf = sparkSession.read.option("header","true").option("inferSchema","true").csv("Person.txt").as(encoder)

  personDf.printSchema()
  personDf.show()

  val calAge = personDf.map(p => {
    p.age = Year.now().getValue - p.dob.substring(6).toInt
    println(p.age)
    p
  } )//.toDF()//.as(encoder)

  print("*********Person DF Schema after age calculation: ")
  calAge.printSchema()

  //calAge.show
}

【问题讨论】：

一个 case 类 不是 java bean。您只需要这样做：import sparkSession.implcits._，然后是 sparkSession.read.option("header","true").option("inferSchema","true").csv("Person.txt").as[Person]，这在 getting started page of the documentation 中进行了解释 - 此外，不鼓励在 map 中使用 print，并且在真正的分布式部署中不会按预期工作 -最后，case classes 应该是 final - 在做之前花点时间阅读和学习会更好。

标签： scala apache-spark apache-spark-sql apache-spark-encoders

【解决方案1】：

package spark

import java.text.SimpleDateFormat
import java.util.Calendar

import org.apache.spark.sql.{SparkSession}
import org.apache.spark.sql.functions._

case class Person(firstName: String, lastName: String, dob: String, age: Long)

object CalcAge extends App {

  val spark = SparkSession.builder()
    .master("local")
    .appName("DataFrame-example")
    .getOrCreate()

  import spark.implicits._

  val sourceDF = Seq(
    ("ABC", "XYZ", "01/01/2019"),
    ("CDE", "FGH", "01/02/2020")
  ).toDF("firstName","lastName","dob")

  sourceDF.printSchema
  //  root
  //  |-- firstName: string (nullable = true)
  //  |-- lastName: string (nullable = true)
  //  |-- dob: string (nullable = true)

  sourceDF.show(false)
  //  +---------+--------+----------+
  //  |firstName|lastName|dob       |
  //  +---------+--------+----------+
  //  |ABC      |XYZ     |01/01/2019|
  //  |CDE      |FGH     |01/02/2020|
  //  +---------+--------+----------+


  def getCurrentYear: Long = {

    val today:java.util.Date = Calendar.getInstance.getTime
    val timeFormat = new SimpleDateFormat("yyyy")
    timeFormat.format(today).toLong

  }

  val ageUDF = udf((d1: String) => {

    val year = d1.split("/").reverse.head.toLong
    val yearNow = getCurrentYear
    yearNow - year
  })


  val df = sourceDF
    .withColumn("age", ageUDF('dob))
  df.printSchema
  //  root
  //  |-- firstName: string (nullable = true)
  //  |-- lastName: string (nullable = true)
  //  |-- dob: string (nullable = true)
  //  |-- age: long (nullable = false)

  df.show(false)
  //  +---------+--------+----------+---+
  //  |firstName|lastName|dob       |age|
  //  +---------+--------+----------+---+
  //  |ABC      |XYZ     |01/01/2019|1  |
  //  |CDE      |FGH     |01/02/2020|0  |
  //  +---------+--------+----------+---+

  val person = df.as[Person].collectAsList()
  //  person: java.util.List[Person] = [Person(ABC,XYZ,01/01/2019,1), Person(CDE,FGH,01/02/2020,0)]
  println(person)



}

【讨论】：