将形式为"MM/dd/yyyy HH:mm"的字符串转换为Spark中数据框中的joda datetime

Question

我在csv文件中读取一列中应该转换为日期时间的字符串.字符串在表单中MM/dd/yyyy HH:mm.但是当我尝试使用joda-time转换它时,我总是得到错误:

线程"main"中的异常java.lang.UnsupportedOperationException:不支持类型为org.joda.time.DateTime的模式

我不知道究竟是什么问题......

 val input = c.textFile("C:\\Users\\AAPL.csv").map(_.split(",")).map{p => 
      val formatter: DateTimeFormatter = DateTimeFormat.forPattern("MM/dd/yyyy HH:mm");
      val date: DateTime = formatter.parseDateTime(p(0));
      StockData(date, p(1).toDouble, p(2).toDouble, p(3).toDouble, p(4).toDouble, p(5).toInt, p(6).toInt)
}.toDF()

谁能帮忙？

Answer 1

我不知道究竟是什么问题......

好吧,问题的根源几乎是由错误消息描述的.Spark SQL不支持Joda-Time DateTime作为输入.日期字段的有效输入是java.sql.Date(请参阅Spark SQL和DataFrame指南,数据类型以供参考).

最简单的解决方案是调整StockData类,以便将其java.sql.Data作为参数并替换:

val date: DateTime = formatter.parseDateTime(p(0))

用这样的东西:

val date: java.sql.Date = new java.sql.Date(
  formatter.parseDateTime(p(0)).getMillis)

要么

val date: java.sql.Timestamp = new java.sql.Timestamp(
  formatter.parseDateTime(p(0)).getMillis)

如果你想保留小时/分钟.

如果您考虑使用带有range子句的窗口函数,更好的选择是将字符串传递给DataFrame并将其转换为整数时间戳:

import org.apache.spark.sql.functions.unix_timestamp

df.withColumn("ts", unix_timestamp($"date", "MM/dd/yyyy HH:mm"))

有关详细信息,请参阅日期之间的Spark窗口函数 - rangeBet.