将带有joda.DateTime的案例类的DStream转换为Spark DataFrame

Posted 2023-04-17

【中文标题】将带有joda.DateTime的案例类的DStream转换为Spark DataFrame

【英文标题】：Convert DStream of case class with joda.DateTime to Spark DataFrame

【发布时间】：2016-09-22 11:01:40

【问题描述】：

我想使用 parquet 格式将 DStream 保存到 HDFS 中。问题是我的案例类使用 joda.DateTime 而 Spark SQL 不支持这个。例如：

    case class Log (timestamp: DateTime, ...dozen of other fields here...)

但我收到错误：java.lang.UnsupportedOperationException：尝试将 RDD 转换为 DF 时不支持 org.joda.time.DateTime 类型的架构：

    def output(logdstream: DStream[Log]) 
        logdstream.foreachRDD(elem => 
           val df = elem.toDF()
           df.saveAsParquet(...)
        );
    

我的模型很复杂并且有很多字段，所以我不想编写不同的案例类来摆脱 joda.DateTime。另一种选择是直接从 json 保存到镶木地板，但这并不理想。有没有一种简单的方法可以将 joda.DateTime 自动转换为 sql.Timestamp 以与 spark 一起使用（转换为 Spark 的数据帧）。

谢谢。

【参考方案1】：

有点冗长，但您可以尝试将 Log 映射到 Spark SQL Row：

logdstream.foreachRDD(rdd => 
  rdd.map(log => Row(
    log.timestamp.toDate,
    log.field2,
    ...
  )).toDF().saveAsParquest(...)
)

【讨论】：

嗨，不确定我是否理解正确。但是错误发生在语句中： val df = elem.toDF();换句话说，我无法使用 .toDF() 函数将 RDD[Log] 转换为数据帧。您建议的解决方案似乎假设 df 已经可用？

你说得对，我错过了。我已经改变了答案。