Spark -Scala - 将 CSV 文件转换为自定义对象

Question

如何将 csv 数据转换为 spark 中的自定义对象。下面是我的代码sn-p

val sparkSession = SparkSession
      .builder()
      .appName("Spark SQL basic example")
      .master("local[2]")
      .getOrCreate()

    val citiData = sparkSession.read.option("header", "true").option("inferSchema", "true").csv(filePath) // removing header,and applying schema

    //citiData.describe().show()
    import sparkSession.implicits._
    val s: Dataset[CityData] = citiData.as[CityData]

  }
  //Date,Open,High,Low,Close,Volume
  case class CityData(processingDate: java.util.Date, Open: Double, High: Double, Low: Double, Volume: Double)

示例数据集：

Date,Open,High,Low,Close,Volume
2006-01-03,490.0,493.8,481.1,492.9,1537660
2006-01-04,488.6,491.0,483.5,483.8,1871020
2006-01-05,484.4,487.8,484.0,486.2,1143160
2006-01-06,488.8,489.0,482.0,486.2,1370250

我已将案例类 CityData 输入参数类型更改为 String ，然后导致“无法解析 'processingDate' 给定输入列：[Volume, Close, High, Date, Low, Open];”例外。

1. 如何创建自定义对象
2. 转换为 Date 对象的另一个棘手问题

我该怎么办？请分享您的想法。

Answer 1

在您的情况下，如果您没有将选项 header 设置为 true，Spark 将读取 String 类型的列。使用选项header，可以看到；

val df = sqlContext.read.option("header", true).option("inferSchema", true).csv("pathToFile")
df.printSchema
//Prints
root
|-- Date: timestamp (nullable = true)
|-- Open: double (nullable = true)
|-- High: double (nullable = true)
|-- Low: double (nullable = true)
|-- Close: double (nullable = true)
|-- Volume: integer (nullable = true)

如果你尝试将行转换为CityData，你会得到如下错误；

java.lang.UnsupportedOperationException: No Encoder found for java.util.Date

这意味着，您不能将TimestampType 直接转换为java.util.Date。这是类型映射；

• TimestampType => java.sql.Timestamp
• DateType => java.sql.Date

将processingDate 的类型从java.util.Date 更改为java.sql.Timestamp 后，您仍然会收到一条错误消息，显示为cannot resolve 'processingDate'。您还需要将CityData 中的字段名称processingDate 更改为Date。然后您可以使用df.as[CityData] 将您的数据集转换为Dataset[CityData]。希望对你有帮助！